From c3cd7a68f6b9b02e9746cdaab2cd4be545adcc4c Mon Sep 17 00:00:00 2001 From: Adil Date: Sun, 30 Aug 2026 23:58:15 +0500 Subject: [PATCH 1/7] chore(ids): move Hid/HidPrefix into ids, re-export from ensemble Move the content-addressable identity primitive (Hid, HidPrefix, HidParseError) from ensemble into ids so light bee/identity crates can mint Hids without dragging the whole mesh (iroh/rustls/etc). ensemble re-exports ids::{Hid,HidPrefix,HidParseError} for back-compat, so existing ensemble::Hid call sites keep compiling. ids gains a hex dep for the wire-form encoding. --- Cargo.lock | 1 + ensemble/src/lib.rs | 173 +------------------------------------------- ids/Cargo.toml | 1 + ids/src/lib.rs | 172 +++++++++++++++++++++++++++++++++++++++++++ 4 files changed, 178 insertions(+), 169 deletions(-) diff --git a/Cargo.lock b/Cargo.lock index 393d889..c6e381c 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -2231,6 +2231,7 @@ dependencies = [ name = "ids" version = "0.32.0" dependencies = [ + "hex", "rand 0.8.6", "serde", "serde_json", diff --git a/ensemble/src/lib.rs b/ensemble/src/lib.rs index f09ce7e..0a75d62 100644 --- a/ensemble/src/lib.rs +++ b/ensemble/src/lib.rs @@ -34,9 +34,7 @@ use anyhow::Result; use async_trait::async_trait; use ed25519_dalek::{Signature, Signer, SigningKey, Verifier, VerifyingKey}; use parking_lot::RwLock; -use rand::RngCore; use serde::{Deserialize, Serialize}; -use sha2::{Digest, Sha256}; use tokio::sync::{broadcast, mpsc}; use tokio::task::JoinSet; @@ -89,173 +87,10 @@ const HANDSHAKE_SKEW_MS: i64 = 60_000; pub type Tone = serde_json::Value; // ── Identity ─────────────────────────────────────────────────────────────── - -/// Universal identity primitive for everything addressable on the -/// hum wire: humds, worker bees, forager bees, future kinds. 32-byte -/// SHA-256 of a public key (Ed25519); wire form is `_` -/// where prefix discriminates the role. -/// -/// Short form keeps the prefix: `humd_a4f2b8c19d3e` (12 hex chars -/// after the underscore). Long form is the full 64-hex tail. Both -/// parse via [`Hid::from_str`]. -/// -/// Prefixes locked in v0: -/// - `humd_` — daemon -/// - `wbee_` — worker bee -/// - `fbee_` — forager bee -/// -/// Stable per install (each binary persists its own key). The hex is -/// content-addressable — no registry, no central naming. -#[derive(Debug, Clone, Copy, PartialEq, Eq, Hash)] -pub struct Hid { - pub prefix: HidPrefix, - pub bytes: [u8; 32], -} - -#[derive(Debug, Clone, Copy, PartialEq, Eq, Hash)] -pub enum HidPrefix { - Humd, - Wbee, - Fbee, -} - -impl HidPrefix { - pub fn as_str(self) -> &'static str { - match self { - HidPrefix::Humd => "humd", - HidPrefix::Wbee => "wbee", - HidPrefix::Fbee => "fbee", - } - } - pub fn parse(s: &str) -> Option { - match s { - "humd" => Some(HidPrefix::Humd), - "wbee" => Some(HidPrefix::Wbee), - "fbee" => Some(HidPrefix::Fbee), - _ => None, - } - } -} - -impl Hid { - /// Hash a pubkey to its content-addressable bytes; tag with the - /// role prefix. - pub fn from_pubkey(prefix: HidPrefix, pubkey: &[u8]) -> Self { - let mut h = Sha256::new(); - h.update(pubkey); - let digest = h.finalize(); - let mut bytes = [0u8; 32]; - bytes.copy_from_slice(&digest[..32]); - Self { prefix, bytes } - } - - /// Mint a random hid for the given role. Tests / pre-crypto only. - pub fn random(prefix: HidPrefix) -> Self { - let mut bytes = [0u8; 32]; - rand::thread_rng().fill_bytes(&mut bytes); - Self { prefix, bytes } - } - - /// Convenience: random hid tagged as a humd. Matches the legacy - /// `Hid::random_humd()` shape; preferred call sites use - /// [`Hid::random`] directly. - pub fn random_humd() -> Self { Self::random(HidPrefix::Humd) } - - pub fn as_bytes(&self) -> &[u8; 32] { &self.bytes } - - /// Full wire form: `_<64 hex>`. - pub fn to_hex(&self) -> String { - format!("{}_{}", self.prefix.as_str(), hex::encode(self.bytes)) - } - - /// Log-friendly short form: `_<12 hex>`. - pub fn short(&self) -> String { - format!("{}_{}", self.prefix.as_str(), hex::encode(&self.bytes[..6])) - } - - /// Parse `_`. Accepts either the 12-char short or - /// 64-char full form. Also accepts a bare 64-hex string as a - /// humd-prefixed legacy value so old peers.json keeps loading. - pub fn from_hex(s: &str) -> Result { - if let Some((p, h)) = s.split_once('_') { - let prefix = HidPrefix::parse(p).ok_or(HidParseError::UnknownPrefix)?; - let tail = hex::decode(h).map_err(|_| HidParseError::BadHex)?; - if tail.len() == 32 { - let mut bytes = [0u8; 32]; - bytes.copy_from_slice(&tail); - return Ok(Hid { prefix, bytes }); - } - // 12-char short form: 6 bytes; widen with zeros so the - // short can still round-trip through serializers (used in - // logs + sigils, not for security-bearing addresses). - if tail.len() == 6 { - let mut bytes = [0u8; 32]; - bytes[..6].copy_from_slice(&tail); - return Ok(Hid { prefix, bytes }); - } - return Err(HidParseError::WrongLength(tail.len())); - } - // Legacy fallback: bare 64-hex was the daemon's old wire form. - // Auto-prefix as `humd_` so older peers.json keeps parsing. - let bytes_vec = hex::decode(s).map_err(|_| HidParseError::BadHex)?; - if bytes_vec.len() != 32 { - return Err(HidParseError::WrongLength(bytes_vec.len())); - } - let mut bytes = [0u8; 32]; - bytes.copy_from_slice(&bytes_vec); - Ok(Hid { prefix: HidPrefix::Humd, bytes }) - } -} - -#[derive(Debug)] -pub enum HidParseError { - UnknownPrefix, - BadHex, - WrongLength(usize), -} - -impl fmt::Display for HidParseError { - fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result { - match self { - HidParseError::UnknownPrefix => write!(f, "unknown hid prefix"), - HidParseError::BadHex => write!(f, "bad hex"), - HidParseError::WrongLength(n) => write!(f, "wrong hid length ({n} bytes)"), - } - } -} - -impl std::error::Error for HidParseError {} - -impl fmt::Display for Hid { - fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result { - write!(f, "{}", self.to_hex()) - } -} - -impl Serialize for Hid { - fn serialize(&self, ser: S) -> Result - where S: serde::Serializer { - ser.serialize_str(&self.to_hex()) - } -} - -impl<'de> Deserialize<'de> for Hid { - fn deserialize(de: D) -> Result - where D: serde::Deserializer<'de> { - let s = String::deserialize(de)?; - Hid::from_hex(&s).map_err(serde::de::Error::custom) - } -} - -impl From<[u8; 32]> for Hid { - /// Legacy: bare 32-byte construction defaults to `humd_` prefix. - /// Migrate to `Hid { prefix, bytes }` directly when the role is - /// known. - fn from(bytes: [u8; 32]) -> Self { - Self { prefix: HidPrefix::Humd, bytes } - } -} - +/// Content-addressable identity, moved to `ids`. Re-exported here for +/// back-compat so existing `ensemble::Hid` / `ensemble::HidPrefix` call +/// sites keep compiling. +pub use ids::{Hid, HidPrefix, HidParseError}; /// Ed25519 signing key for a humd. The pubkey's SHA-256 is the /// [`Hid`] — identity is content-addressable, no separate registry. /// diff --git a/ids/Cargo.toml b/ids/Cargo.toml index 836beae..1aa6f82 100644 --- a/ids/Cargo.toml +++ b/ids/Cargo.toml @@ -9,6 +9,7 @@ description = "hum-native 256-bit Crockford-base32 identifiers: 48-bit ms timest rand = { workspace = true } serde = { workspace = true, features = ["derive"] } sha2 = { workspace = true } +hex = { workspace = true } uuid = { version = "1", features = ["v5"] } [dev-dependencies] diff --git a/ids/src/lib.rs b/ids/src/lib.rs index eae51b2..88fc942 100644 --- a/ids/src/lib.rs +++ b/ids/src/lib.rs @@ -318,3 +318,175 @@ mod tests { assert!(!is_valid_id(&short)); } } + +// ── Hid ──────────────────────────────────────────────────────────────────── + +/// Universal identity primitive for everything addressable on the +/// hum wire: humds, worker bees, forager bees, future kinds. 32-byte +/// SHA-256 of a public key (Ed25519); wire form is `_` +/// where prefix discriminates the role. +/// +/// Short form keeps the prefix: `humd_a4f2b8c19d3e` (12 hex chars +/// after the underscore). Long form is the full 64-hex tail. Both +/// parse via [`Hid::from_hex`]. +/// +/// Prefixes locked in v0: +/// - `humd_` — daemon +/// - `wbee_` — worker bee +/// - `fbee_` — forager bee +/// +/// Stable per install (each binary persists its own key). The hex is +/// content-addressable — no registry, no central naming. +/// +/// Originally defined in `ensemble`; moved here so light bee/identity +/// crates can mint [`Hid`]s without dragging the whole mesh. `ensemble` +/// re-exports it for back-compat. +#[derive(Debug, Clone, Copy, PartialEq, Eq, Hash)] +pub struct Hid { + pub prefix: HidPrefix, + pub bytes: [u8; 32], +} + +#[derive(Debug, Clone, Copy, PartialEq, Eq, Hash)] +pub enum HidPrefix { + Humd, + Wbee, + Fbee, +} + +impl HidPrefix { + pub fn as_str(self) -> &'static str { + match self { + HidPrefix::Humd => "humd", + HidPrefix::Wbee => "wbee", + HidPrefix::Fbee => "fbee", + } + } + pub fn parse(s: &str) -> Option { + match s { + "humd" => Some(HidPrefix::Humd), + "wbee" => Some(HidPrefix::Wbee), + "fbee" => Some(HidPrefix::Fbee), + _ => None, + } + } +} + +impl Hid { + /// Hash a pubkey to its content-addressable bytes; tag with the + /// role prefix. + pub fn from_pubkey(prefix: HidPrefix, pubkey: &[u8]) -> Self { + let mut h = Sha256::new(); + h.update(pubkey); + let digest = h.finalize(); + let mut bytes = [0u8; 32]; + bytes.copy_from_slice(&digest[..32]); + Self { prefix, bytes } + } + + /// Mint a random hid for the given role. Tests / pre-crypto only. + pub fn random(prefix: HidPrefix) -> Self { + let mut bytes = [0u8; 32]; + rand::thread_rng().fill_bytes(&mut bytes); + Self { prefix, bytes } + } + + /// Convenience: random hid tagged as a humd. Matches the legacy + /// `Hid::random_humd()` shape; preferred call sites use + /// [`Hid::random`] directly. + pub fn random_humd() -> Self { Self::random(HidPrefix::Humd) } + + pub fn as_bytes(&self) -> &[u8; 32] { &self.bytes } + + /// Full wire form: `_<64 hex>`. + pub fn to_hex(&self) -> String { + format!("{}_{}", self.prefix.as_str(), hex::encode(self.bytes)) + } + + /// Log-friendly short form: `_<12 hex>`. + pub fn short(&self) -> String { + format!("{}_{}", self.prefix.as_str(), hex::encode(&self.bytes[..6])) + } + + /// Parse `_`. Accepts either the 12-char short or + /// 64-char full form. Also accepts a bare 64-hex string as a + /// humd-prefixed legacy value so old peers.json keeps loading. + pub fn from_hex(s: &str) -> Result { + if let Some((p, h)) = s.split_once('_') { + let prefix = HidPrefix::parse(p).ok_or(HidParseError::UnknownPrefix)?; + let tail = hex::decode(h).map_err(|_| HidParseError::BadHex)?; + if tail.len() == 32 { + let mut bytes = [0u8; 32]; + bytes.copy_from_slice(&tail); + return Ok(Hid { prefix, bytes }); + } + // 12-char short form: 6 bytes; widen with zeros so the + // short can still round-trip through serializers (used in + // logs + sigils, not for security-bearing addresses). + if tail.len() == 6 { + let mut bytes = [0u8; 32]; + bytes[..6].copy_from_slice(&tail); + return Ok(Hid { prefix, bytes }); + } + return Err(HidParseError::WrongLength(tail.len())); + } + // Legacy fallback: bare 64-hex was the daemon's old wire form. + // Auto-prefix as `humd_` so older peers.json keeps parsing. + let bytes_vec = hex::decode(s).map_err(|_| HidParseError::BadHex)?; + if bytes_vec.len() != 32 { + return Err(HidParseError::WrongLength(bytes_vec.len())); + } + let mut bytes = [0u8; 32]; + bytes.copy_from_slice(&bytes_vec); + Ok(Hid { prefix: HidPrefix::Humd, bytes }) + } +} + +#[derive(Debug)] +pub enum HidParseError { + UnknownPrefix, + BadHex, + WrongLength(usize), +} + +impl fmt::Display for HidParseError { + fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result { + match self { + HidParseError::UnknownPrefix => write!(f, "unknown hid prefix"), + HidParseError::BadHex => write!(f, "bad hex"), + HidParseError::WrongLength(n) => write!(f, "wrong hid length ({n} bytes)"), + } + } +} + +impl std::error::Error for HidParseError {} + +impl fmt::Display for Hid { + fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result { + write!(f, "{}", self.to_hex()) + } +} + +impl Serialize for Hid { + fn serialize(&self, ser: S) -> Result + where S: serde::Serializer { + ser.serialize_str(&self.to_hex()) + } +} + +impl<'de> Deserialize<'de> for Hid { + fn deserialize(de: D) -> Result + where D: serde::Deserializer<'de> { + let s = String::deserialize(de)?; + Hid::from_hex(&s).map_err(serde::de::Error::custom) + } +} + +impl From<[u8; 32]> for Hid { + /// Legacy: bare 32-byte construction defaults to `humd_` prefix. + /// Migrate to `Hid { prefix, bytes }` directly when the role is + /// known. + fn from(bytes: [u8; 32]) -> Self { + Self { prefix: HidPrefix::Humd, bytes } + } +} From 6bbe4ec40460811208400a51ec16e3d21c357c4c Mon Sep 17 00:00:00 2001 From: Adil Date: Mon, 31 Aug 2026 00:07:47 +0500 Subject: [PATCH 2/7] feat(hum-identity): carve persistent bee identity into a leaf crate Extract load_or_mint_bee_key / BeeKey / bee_key_path from nest-common into a standalone hum-identity crate so remote hives can depend on it without pulling the daemon tree (ensemble/iroh, nest, mcp, etc). - New crate hum-identity: ids (Hid) + hum-paths + ed25519-dalek + rand. - hives/common/src/identity.rs becomes a thin re-export shim for back-compat; existing nest_common::load_or_mint_bee_key call sites keep resolving. - Registered in workspace members; added to hives/common deps. --- Cargo.lock | 14 +++ Cargo.toml | 1 + hives/common/Cargo.toml | 1 + hives/common/src/identity.rs | 165 +++-------------------------------- hum-identity/Cargo.toml | 17 ++++ hum-identity/src/lib.rs | 163 ++++++++++++++++++++++++++++++++++ 6 files changed, 206 insertions(+), 155 deletions(-) create mode 100644 hum-identity/Cargo.toml create mode 100644 hum-identity/src/lib.rs diff --git a/Cargo.lock b/Cargo.lock index c6e381c..5ec5777 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -1915,6 +1915,19 @@ dependencies = [ "tracing-subscriber", ] +[[package]] +name = "hum-identity" +version = "0.32.0" +dependencies = [ + "anyhow", + "ed25519-dalek 2.2.0", + "hum-paths", + "ids", + "rand 0.8.6", + "tempfile", + "tracing", +] + [[package]] name = "hum-paths" version = "0.32.0" @@ -2987,6 +3000,7 @@ dependencies = [ "ed25519-dalek 2.2.0", "ensemble", "futures", + "hum-identity", "hum-paths", "ids", "lru 0.12.5", diff --git a/Cargo.toml b/Cargo.toml index aed08f9..63f96ee 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -4,6 +4,7 @@ members = [ "thrum-core", "thrumd", "hum-paths", + "hum-identity", "ids", "config", "codegen", diff --git a/hives/common/Cargo.toml b/hives/common/Cargo.toml index 8494ca9..79d19d2 100644 --- a/hives/common/Cargo.toml +++ b/hives/common/Cargo.toml @@ -7,6 +7,7 @@ description = "Shared building blocks for nests — regex Classifier for the dro [dependencies] hum-paths = { path = "../../hum-paths" } +hum-identity = { path = "../../hum-identity" } drone = { path = "../../drone" } ensemble = { path = "../../ensemble" } mcp = { path = "../../mcp" } diff --git a/hives/common/src/identity.rs b/hives/common/src/identity.rs index 3efc1bf..b218228 100644 --- a/hives/common/src/identity.rs +++ b/hives/common/src/identity.rs @@ -1,158 +1,13 @@ -//! Persistent bee identity. Mirrors humd's `identity.rs` but tags -//! the resulting [`Hid`] with the bee role (worker / forager). +//! Bee identity — now a leaf crate. //! -//! Each bee install gets its own Ed25519 keypair at -//! `$XDG_STATE_HOME/hum/bees/.key`. The pubkey hashes to a -//! stable role-prefixed [`Hid`] (`wbee_` / `fbee_`) that -//! survives reconnect, restart, even daemon swap. The bee carries -//! this hid in every `chi:"hello"` so humd's manifest registry is -//! keyed by identity, not by transient thrum connection id. +//! The persistent-identity logic (load-or-mint an Ed25519 seed at +//! `$XDG_STATE_HOME/hum/bees/.key`, derive the role-tagged +//! [`Hid`]) lives in the standalone `hum-identity` crate, so remote +//! hives can depend on it without pulling in the daemon tree. //! -//! File format: raw 32-byte Ed25519 secret seed, mode 0o600, atomic -//! write-and-rename. Same convention as humd's daemon key — the only -//! difference is the dir + role tagging. +//! This module is a thin re-export shim for back-compat during the +//! transition: existing `nest_common::load_or_mint_bee_key` / +//! `nest_common::BeeKey` / `nest_common::bee_key_path` call sites keep +//! resolving. New hives should depend on `hum-identity` directly. -use std::fs; -use std::io::Write; -use std::path::{Path, PathBuf}; - -use anyhow::{anyhow, Context, Result}; -use ed25519_dalek::SigningKey; -use ensemble::{Hid, HidPrefix}; -use rand::RngCore; -use tracing::{info, trace}; - -/// One bee's persistent identity: signing key + derived [`Hid`]. -/// Cheap to clone the bytes; the signing key is on the stack inside -/// the wrapper. -#[derive(Debug)] -pub struct BeeKey { - pub signing: SigningKey, - pub hid: Hid, -} - -impl BeeKey { - pub fn pubkey_bytes(&self) -> [u8; 32] { - self.signing.verifying_key().to_bytes() - } -} - -pub fn bee_key_path(kind: &str) -> PathBuf { - hum_paths::bee_key(kind) -} - -/// Load the bee's persisted key, minting + persisting a fresh one -/// on first boot. The returned [`Hid`] is derived from the pubkey -/// with the given role `prefix`. -pub fn load_or_mint_bee_key(kind: &str, prefix: HidPrefix) -> Result { - let path = bee_key_path(kind); - if path.exists() { - let bytes = fs::read(&path) - .with_context(|| format!("read bee key {}", path.display()))?; - if bytes.len() != 32 { - return Err(anyhow!( - "bee key at {} is {} bytes, expected 32", - path.display(), - bytes.len() - )); - } - let mut arr = [0u8; 32]; - arr.copy_from_slice(&bytes); - let signing = SigningKey::from_bytes(&arr); - let hid = Hid::from_pubkey(prefix, &signing.verifying_key().to_bytes()); - trace!(path = %path.display(), %kind, hid = %hid.short(), "bee.identity.loaded"); - return Ok(BeeKey { signing, hid }); - } - - let mut seed = [0u8; 32]; - rand::thread_rng().fill_bytes(&mut seed); - let signing = SigningKey::from_bytes(&seed); - persist(&path, &seed)?; - let hid = Hid::from_pubkey(prefix, &signing.verifying_key().to_bytes()); - info!(path = %path.display(), %kind, hid = %hid.short(), "bee.identity.minted"); - Ok(BeeKey { signing, hid }) -} - -fn persist(path: &Path, seed: &[u8; 32]) -> Result<()> { - if let Some(parent) = path.parent() { - fs::create_dir_all(parent) - .with_context(|| format!("mkdir -p {}", parent.display()))?; - } - let tmp = match path.file_name() { - Some(name) => { - let mut tmp_name = name.to_os_string(); - tmp_name.push(".tmp"); - path.with_file_name(tmp_name) - } - None => return Err(anyhow!("bee key path has no file name: {}", path.display())), - }; - - { - let mut opts = fs::OpenOptions::new(); - opts.write(true).create(true).truncate(true); - #[cfg(unix)] - { - use std::os::unix::fs::OpenOptionsExt; - opts.mode(0o600); - } - let mut f = opts - .open(&tmp) - .with_context(|| format!("create {}", tmp.display()))?; - f.write_all(seed) - .with_context(|| format!("write {}", tmp.display()))?; - f.sync_all().ok(); - } - - #[cfg(unix)] - { - use std::os::unix::fs::PermissionsExt; - let _ = fs::set_permissions(&tmp, fs::Permissions::from_mode(0o600)); - } - - fs::rename(&tmp, path) - .with_context(|| format!("rename {} -> {}", tmp.display(), path.display()))?; - Ok(()) -} - -#[cfg(test)] -mod tests { - use super::*; - use std::sync::Mutex; - use tempfile::TempDir; - - // Serialize tests because they share the XDG_STATE_HOME env var. - // Without this lock, parallel cargo-test threads race the env and - // the second test sees the first's tempdir (or an empty value). - static ENV_LOCK: Mutex<()> = Mutex::new(()); - - #[test] - fn round_trip_worker_key_then_different_kind() { - let _guard = ENV_LOCK.lock().unwrap(); - let tmp = TempDir::new().unwrap(); - std::env::set_var("XDG_STATE_HOME", tmp.path()); - - let first = load_or_mint_bee_key("claude-cli", HidPrefix::Wbee).expect("mint"); - assert_eq!(first.hid.prefix, HidPrefix::Wbee); - let id1 = first.hid; - - let second = load_or_mint_bee_key("claude-cli", HidPrefix::Wbee).expect("reload"); - assert_eq!(id1, second.hid, "wbee hid stable across reloads"); - - let other = load_or_mint_bee_key("humfs", HidPrefix::Fbee).expect("mint"); - assert_ne!(id1, other.hid); - assert_eq!(other.hid.prefix, HidPrefix::Fbee); - - std::env::remove_var("XDG_STATE_HOME"); - } - - #[test] - fn key_path_uses_xdg_state_home() { - let _guard = ENV_LOCK.lock().unwrap(); - let tmp = TempDir::new().unwrap(); - std::env::set_var("XDG_STATE_HOME", tmp.path()); - let path = bee_key_path("foo"); - assert!(path.starts_with(tmp.path()), "path {:?}", path); - assert!(path.ends_with("hum/bees/foo.key")); - std::env::remove_var("XDG_STATE_HOME"); - } -} +pub use hum_identity::{bee_key_path, load_or_mint_bee_key, BeeKey}; diff --git a/hum-identity/Cargo.toml b/hum-identity/Cargo.toml new file mode 100644 index 0000000..0b1c0e9 --- /dev/null +++ b/hum-identity/Cargo.toml @@ -0,0 +1,17 @@ +[package] +name = "hum-identity" +version.workspace = true +edition.workspace = true +license.workspace = true +description = "Persistent bee identity: load-or-mint an Ed25519 seed and derive the role-tagged Hid. Leaf crate for remote hives — no daemon, mesh, or nest deps." + +[dependencies] +ids = { path = "../ids" } +hum-paths = { path = "../hum-paths" } +ed25519-dalek = { version = "2", features = ["rand_core"] } +rand = { workspace = true } +anyhow = { workspace = true } +tracing = { workspace = true } + +[dev-dependencies] +tempfile = "3" diff --git a/hum-identity/src/lib.rs b/hum-identity/src/lib.rs new file mode 100644 index 0000000..38ee053 --- /dev/null +++ b/hum-identity/src/lib.rs @@ -0,0 +1,163 @@ +//! Persistent bee identity. +//! +//! Mirrors humd's `identity.rs` but tags the resulting [`Hid`] with the +//! bee role (worker / forager). Each bee install gets its own Ed25519 +//! keypair at `$XDG_STATE_HOME/hum/bees/.key`. The pubkey hashes +//! to a stable role-prefixed [`Hid`] (`wbee_` / `fbee_`) that +//! survives reconnect, restart, even daemon swap. The bee carries this +//! hid in every `chi:"hello"` so humd's manifest registry is keyed by +//! identity, not by transient thrum connection id. +//! +//! File format: raw 32-byte Ed25519 secret seed, mode 0o600, atomic +//! write-and-rename. Same convention as humd's daemon key — the only +//! difference is the dir + role tagging. +//! +//! This is a *leaf* crate for remote hives: it depends only on `ids` +//! (for [`Hid`]) and `hum-paths` (for the seed path). No daemon, mesh, +//! or nest machinery. + +use std::fs; +use std::io::Write; +use std::path::{Path, PathBuf}; + +use anyhow::{anyhow, Context, Result}; +use ed25519_dalek::SigningKey; +use ids::{Hid, HidPrefix}; +use rand::RngCore; +use tracing::{info, trace}; + +/// One bee's persistent identity: signing key + derived [`Hid`]. +/// Cheap to clone the bytes; the signing key is on the stack inside +/// the wrapper. +#[derive(Debug)] +pub struct BeeKey { + pub signing: SigningKey, + pub hid: Hid, +} + +impl BeeKey { + pub fn pubkey_bytes(&self) -> [u8; 32] { + self.signing.verifying_key().to_bytes() + } +} + +/// Path to a bee kind's identity seed: `$XDG_STATE_HOME/hum/bees/.key`. +pub fn bee_key_path(kind: &str) -> PathBuf { + hum_paths::bee_key(kind) +} + +/// Load the bee's persisted key, minting + persisting a fresh one +/// on first boot. The returned [`Hid`] is derived from the pubkey +/// with the given role `prefix`. +pub fn load_or_mint_bee_key(kind: &str, prefix: HidPrefix) -> Result { + let path = bee_key_path(kind); + if path.exists() { + let bytes = fs::read(&path) + .with_context(|| format!("read bee key {}", path.display()))?; + if bytes.len() != 32 { + return Err(anyhow!( + "bee key at {} is {} bytes, expected 32", + path.display(), + bytes.len() + )); + } + let mut arr = [0u8; 32]; + arr.copy_from_slice(&bytes); + let signing = SigningKey::from_bytes(&arr); + let hid = Hid::from_pubkey(prefix, &signing.verifying_key().to_bytes()); + trace!(path = %path.display(), %kind, hid = %hid.short(), "bee.identity.loaded"); + return Ok(BeeKey { signing, hid }); + } + + let mut seed = [0u8; 32]; + rand::thread_rng().fill_bytes(&mut seed); + let signing = SigningKey::from_bytes(&seed); + persist(&path, &seed)?; + let hid = Hid::from_pubkey(prefix, &signing.verifying_key().to_bytes()); + info!(path = %path.display(), %kind, hid = %hid.short(), "bee.identity.minted"); + Ok(BeeKey { signing, hid }) +} + +fn persist(path: &Path, seed: &[u8; 32]) -> Result<()> { + if let Some(parent) = path.parent() { + fs::create_dir_all(parent) + .with_context(|| format!("mkdir -p {}", parent.display()))?; + } + let tmp = match path.file_name() { + Some(name) => { + let mut tmp_name = name.to_os_string(); + tmp_name.push(".tmp"); + path.with_file_name(tmp_name) + } + None => return Err(anyhow!("bee key path has no file name: {}", path.display())), + }; + + { + let mut opts = fs::OpenOptions::new(); + opts.write(true).create(true).truncate(true); + #[cfg(unix)] + { + use std::os::unix::fs::OpenOptionsExt; + opts.mode(0o600); + } + let mut f = opts + .open(&tmp) + .with_context(|| format!("create {}", tmp.display()))?; + f.write_all(seed) + .with_context(|| format!("write {}", tmp.display()))?; + f.sync_all().ok(); + } + + #[cfg(unix)] + { + use std::os::unix::fs::PermissionsExt; + let _ = fs::set_permissions(&tmp, fs::Permissions::from_mode(0o600)); + } + + fs::rename(&tmp, path) + .with_context(|| format!("rename {} -> {}", tmp.display(), path.display()))?; + Ok(()) +} + +#[cfg(test)] +mod tests { + use super::*; + use std::sync::Mutex; + use tempfile::TempDir; + + // Serialize tests because they share the XDG_STATE_HOME env var. + // Without this lock, parallel cargo-test threads race the env and + // the second test sees the first's tempdir (or an empty value). + static ENV_LOCK: Mutex<()> = Mutex::new(()); + + #[test] + fn round_trip_worker_key_then_different_kind() { + let _guard = ENV_LOCK.lock().unwrap(); + let tmp = TempDir::new().unwrap(); + std::env::set_var("XDG_STATE_HOME", tmp.path()); + + let first = load_or_mint_bee_key("claude-cli", HidPrefix::Wbee).expect("mint"); + assert_eq!(first.hid.prefix, HidPrefix::Wbee); + let id1 = first.hid; + + let second = load_or_mint_bee_key("claude-cli", HidPrefix::Wbee).expect("reload"); + assert_eq!(id1, second.hid, "wbee hid stable across reloads"); + + let other = load_or_mint_bee_key("humfs", HidPrefix::Fbee).expect("mint"); + assert_ne!(id1, other.hid); + assert_eq!(other.hid.prefix, HidPrefix::Fbee); + + std::env::remove_var("XDG_STATE_HOME"); + } + + #[test] + fn key_path_uses_xdg_state_home() { + let _guard = ENV_LOCK.lock().unwrap(); + let tmp = TempDir::new().unwrap(); + std::env::set_var("XDG_STATE_HOME", tmp.path()); + let path = bee_key_path("foo"); + assert!(path.starts_with(tmp.path()), "path {:?}", path); + assert!(path.ends_with("hum/bees/foo.key")); + std::env::remove_var("XDG_STATE_HOME"); + } +} From 4e854c73d50d893beb6f3eb940525a5b899e9f46 Mon Sep 17 00:00:00 2001 From: Adil Date: Mon, 31 Aug 2026 00:23:29 +0500 Subject: [PATCH 3/7] feat(hum-thrum): carve the bee wire client into a leaf crate, refactor serve_forager Factor the thrum wire loop (dial Unix socket, split, send NDJSON, read tones, reconnect with jittered backoff) out of nest-common into a standalone hum-thrum crate so remote hives can dial humd without pulling the daemon tree. - New crate hum-thrum: connect / send_json / read_tones / serve_forever. Deps: ids + hum-paths + hum-identity + thrum-core + tokio + serde_json. - serve_forager (forager.rs) now uses hum-thrum for the wire; the ToolDispatcher / ToolDef / ToolResult semantics stay in nest-common. - hum-identity re-exports ids::{Hid,HidPrefix} so consumers get the role type through hum_identity::HidPrefix. - Wire round-trip test: real Unix socket, NDJSON hello contract. Registered in workspace members; added to hives/common deps. --- Cargo.lock | 16 +++++ Cargo.toml | 1 + hives/common/Cargo.toml | 1 + hives/common/src/forager.rs | 45 ++++-------- hum-identity/src/lib.rs | 2 +- hum-thrum/Cargo.toml | 17 +++++ hum-thrum/src/lib.rs | 132 ++++++++++++++++++++++++++++++++++++ 7 files changed, 182 insertions(+), 32 deletions(-) create mode 100644 hum-thrum/Cargo.toml create mode 100644 hum-thrum/src/lib.rs diff --git a/Cargo.lock b/Cargo.lock index 5ec5777..f561170 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -1936,6 +1936,21 @@ dependencies = [ "serde_json", ] +[[package]] +name = "hum-thrum" +version = "0.32.0" +dependencies = [ + "anyhow", + "hum-identity", + "hum-paths", + "ids", + "rand 0.8.6", + "serde_json", + "thrum-core", + "tokio", + "tracing", +] + [[package]] name = "humantime" version = "2.3.0" @@ -3002,6 +3017,7 @@ dependencies = [ "futures", "hum-identity", "hum-paths", + "hum-thrum", "ids", "lru 0.12.5", "mcp", diff --git a/Cargo.toml b/Cargo.toml index 63f96ee..2e0955d 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -5,6 +5,7 @@ members = [ "thrumd", "hum-paths", "hum-identity", + "hum-thrum", "ids", "config", "codegen", diff --git a/hives/common/Cargo.toml b/hives/common/Cargo.toml index 79d19d2..fd7cded 100644 --- a/hives/common/Cargo.toml +++ b/hives/common/Cargo.toml @@ -8,6 +8,7 @@ description = "Shared building blocks for nests — regex Classifier for the dro [dependencies] hum-paths = { path = "../../hum-paths" } hum-identity = { path = "../../hum-identity" } +hum-thrum = { path = "../../hum-thrum" } drone = { path = "../../drone" } ensemble = { path = "../../ensemble" } mcp = { path = "../../mcp" } diff --git a/hives/common/src/forager.rs b/hives/common/src/forager.rs index 7d9a7bd..eb471ee 100644 --- a/hives/common/src/forager.rs +++ b/hives/common/src/forager.rs @@ -18,19 +18,19 @@ //! - **Cancel**: `chi:"cancel"` (with a `callId`) signals the forager //! to abort the in-flight tool, if it can. //! -//! Reconnect is built in — humd restarts don't strand foragers. +//! The wire mechanics (dial, hello, read NDJSON, reconnect) are the +//! shared `hum-thrum` client; this module is the forager-specific +//! chi semantics + tool dispatcher seam. Reconnect is built in — +//! humd restarts don't strand foragers. -use std::path::{Path, PathBuf}; +use std::path::PathBuf; use std::sync::Arc; use anyhow::{Context, Result}; use async_trait::async_trait; -use ensemble::HidPrefix; +use hum_identity::HidPrefix; use serde_json::{json, Value}; -use tokio::io::{AsyncBufReadExt, AsyncWriteExt, BufReader}; -use tokio::net::UnixStream; -use tokio::sync::Mutex; -use tracing::{info, trace, warn}; +use tracing::{info, trace}; use crate::identity::load_or_mint_bee_key; @@ -124,25 +124,16 @@ pub async fn serve_forager( advert: ForagerAdvert, ) -> Result<()> { let path = default_socket_path(); - loop { - match dial_and_serve(&path, dispatcher.clone(), &advert).await { - Ok(()) => trace!("serve_forager: clean exit, reconnecting"), - Err(e) => warn!(err = %e, "serve_forager: connection failed, retrying"), - } - tokio::time::sleep(std::time::Duration::from_secs(2)).await; - } + hum_thrum::serve_forever(|| dial_and_serve(&path, dispatcher.clone(), &advert)).await } async fn dial_and_serve( - path: &Path, + path: &PathBuf, dispatcher: Arc, advert: &ForagerAdvert, ) -> Result<()> { info!(socket = %path.display(), hive = %advert.hive, "forager.connecting"); - let stream = UnixStream::connect(path).await - .with_context(|| format!("connect to thrum at {}", path.display()))?; - let (read_half, write_half) = stream.into_split(); - let write_half = Arc::new(Mutex::new(write_half)); + let (reader, write_half) = hum_thrum::connect(path).await?; // Load (or mint) the persistent forager-bee identity. fbee_ hid // survives reconnect / restart; humd indexes by it. @@ -171,7 +162,7 @@ async fn dial_and_serve( "chis": ["hello", "tool-call", "tool-result", "cancel", "breath", "echo"], "source": advert.source.clone().unwrap_or_default(), }); - write_half.lock().await.write_all(format!("{}\n", hello).as_bytes()).await?; + hum_thrum::send_json(&write_half, &hello).await?; info!( hive = %advert.hive, hid = %bee_key.hid.short(), @@ -180,13 +171,7 @@ async fn dial_and_serve( "forager.hello.sent" ); - let mut reader = BufReader::new(read_half).lines(); - while let Some(line) = reader.next_line().await? { - if line.is_empty() { continue; } - let tone: Value = match serde_json::from_str(&line) { - Ok(v) => v, - Err(e) => { trace!(err = %e, "forager.parse.skip"); continue; } - }; + hum_thrum::read_tones(reader, |tone| { let chi = tone.get("chi").and_then(Value::as_str).unwrap_or(""); match chi { "tool-call" => { @@ -207,13 +192,11 @@ async fn dial_and_serve( "title": result.title, "metadata": result.metadata, }); - let line = format!("{}\n", body); - let _ = write_half.lock().await.write_all(line.as_bytes()).await; + let _ = hum_thrum::send_json(&write_half, &body).await; }); } "breath" | "echo" | "" => {} other => trace!(chi = other, "forager.unknown.chi"), } - } - Ok(()) + }).await } diff --git a/hum-identity/src/lib.rs b/hum-identity/src/lib.rs index 38ee053..89f31d5 100644 --- a/hum-identity/src/lib.rs +++ b/hum-identity/src/lib.rs @@ -22,7 +22,7 @@ use std::path::{Path, PathBuf}; use anyhow::{anyhow, Context, Result}; use ed25519_dalek::SigningKey; -use ids::{Hid, HidPrefix}; +pub use ids::{Hid, HidPrefix}; use rand::RngCore; use tracing::{info, trace}; diff --git a/hum-thrum/Cargo.toml b/hum-thrum/Cargo.toml new file mode 100644 index 0000000..2378de3 --- /dev/null +++ b/hum-thrum/Cargo.toml @@ -0,0 +1,17 @@ +[package] +name = "hum-thrum" +version.workspace = true +edition.workspace = true +license.workspace = true +description = "The thrum wire client for bees: dial the Unix socket, hello, read NDJSON tones, reconnect. Transport-agnostic leaf for remote hives — no daemon, nest, or mesh deps." + +[dependencies] +ids = { path = "../ids" } +hum-paths = { path = "../hum-paths" } +hum-identity = { path = "../hum-identity" } +thrum-core = { path = "../thrum-core" } +tokio = { workspace = true, features = ["net", "io-util", "sync", "macros", "rt", "time"] } +serde_json = { workspace = true } +anyhow = { workspace = true } +tracing = { workspace = true } +rand = { workspace = true } diff --git a/hum-thrum/src/lib.rs b/hum-thrum/src/lib.rs new file mode 100644 index 0000000..993664b --- /dev/null +++ b/hum-thrum/src/lib.rs @@ -0,0 +1,132 @@ +//! `hum-thrum` — the thrum wire client for bees. +//! +//! A bee is a kind + a binary. Its *runtime* only needs the wire: +//! dial the Unix socket humd binds, send a `chi:"hello"`, read NDJSON +//! tones, ship results, and reconnect forever. It does **not** need +//! the daemon's in-memory nest, drone, or ensemble machinery. +//! +//! This crate factors the wire loop out of `serve_worker` / +//! `serve_forager` into a transport-agnostic seam, so a remote hive +//! can depend on it without pulling the daemon tree: +//! +//! - [`connect`] — dial + split the socket, returning a write half. +//! - [`send_json`] — write one NDJSON line (a tone). +//! - [`read_tones`] — read NDJSON lines, parse each into a [`Value`], +//! and hand it to a caller-supplied per-chi dispatcher. +//! - [`serve_forever`] — the reconnect loop with jittered backoff. +//! +//! The chi *semantics* (what a `tool-call`, `prompt`, or `cancel` +//! actually does) stay in the caller — that's worker/forager-specific +//! state. What's shared is the wire mechanics. + +use std::path::Path; +use std::sync::Arc; + +use anyhow::{Context, Result}; +use serde_json::Value; +use tokio::io::{AsyncBufReadExt, AsyncWriteExt, BufReader, Lines}; +use tokio::net::UnixStream; +use tokio::sync::Mutex; +use tracing::{trace, warn}; + +/// Dial the thrum Unix socket and split it. Returns the read half as +/// a line reader and the write half wrapped in a shared mutex (so +/// concurrent spawns can write). +pub async fn connect( + path: &Path, +) -> Result<(Lines>, Arc>)> { + let stream = UnixStream::connect(path).await + .with_context(|| format!("connect to thrum at {}", path.display()))?; + let (read_half, write_half) = stream.into_split(); + let write_half = Arc::new(Mutex::new(write_half)); + let reader = BufReader::new(read_half).lines(); + Ok((reader, write_half)) +} + +/// Write one NDJSON line (a tone) to the shared write half. +pub async fn send_json(write: &Arc>, tone: &Value) -> Result<()> { + let line = format!("{}\n", tone); + write.lock().await.write_all(line.as_bytes()).await?; + Ok(()) +} + +/// Read NDJSON tones and hand each to `on_tone`. Returns `Ok(())` +/// when the stream ends (peer closed — caller reconnects). +pub async fn read_tones(mut reader: Lines>, on_tone: F) -> Result<()> +where + F: Fn(Value), +{ + while let Some(line) = reader.next_line().await? { + if line.is_empty() { continue; } + let tone: Value = match serde_json::from_str(&line) { + Ok(v) => v, + Err(e) => { trace!(err = %e, "thrum.parse.skip"); continue; } + }; + on_tone(tone); + } + Ok(()) +} + +/// The reconnect loop. `dial` is called each attempt (it should +/// connect, hello, and run the tone loop), returning `Ok(())` on a +/// clean exit or `Err` on a failed connection. Sleeps with jittered +/// backoff between attempts so parallel boot races with humd stay +/// quiet (matching serve_worker's grace window). +pub async fn serve_forever(dial: F) -> ! +where + F: Fn() -> Fut, + Fut: std::future::Future>, +{ + let mut consecutive_fails = 0u32; + loop { + match dial().await { + Ok(()) => { + consecutive_fails = 0; + trace!("thrum: clean exit, reconnecting"); + } + Err(e) => { + consecutive_fails += 1; + warn!(err = %e, attempts = consecutive_fails, "thrum: connection failed, retrying"); + } + } + let jitter = rand::random::() * 0.75; + tokio::time::sleep(std::time::Duration::from_secs_f32(2.0 + jitter)).await; + } +} + +#[cfg(test)] +mod tests { + use super::*; + use serde_json::json; + use tokio::net::UnixListener; + use tokio::io::{AsyncBufReadExt, BufReader}; + + // Wire round-trip: a server accepts a Unix connection, the client + // dials via hum-thrum::connect, sends a hello tone, and the server + // reads it back. Confirms connect/send_json/read_tones line up with + // the NDJSON contract humd speaks. + #[tokio::test] + async fn wire_round_trip() { + let dir = std::env::temp_dir().join(format!("hum-thrum-test-{}", std::process::id())); + std::fs::create_dir_all(&dir).unwrap(); + let sock = dir.join("thrum.sock"); + let listener = UnixListener::bind(&sock).unwrap(); + + let server = tokio::spawn(async move { + let (mut stream, _) = listener.accept().await.unwrap(); + let mut buf = String::new(); + let mut r = BufReader::new(&mut stream); + r.read_line(&mut buf).await.unwrap(); + let tone: Value = serde_json::from_str(&buf).unwrap(); + assert_eq!(tone["chi"], "hello"); + assert_eq!(tone["bee"][0], "forager"); + }); + + let (_, write) = connect(&sock).await.unwrap(); + let hello = json!({ "chi": "hello", "bee": ["forager"], "hid": "fbee_abcd" }); + send_json(&write, &hello).await.unwrap(); + + server.await.unwrap(); + std::fs::remove_dir_all(&dir).unwrap(); + } +} From c7a7b6287500454aaa8e594fd69508e8f509aae5 Mon Sep 17 00:00:00 2001 From: Adil Date: Mon, 31 Aug 2026 00:43:02 +0500 Subject: [PATCH 4/7] refactor(serve_worker): use hum-thrum wire primitives in dial_and_serve Swap the worker's manual Unix-socket connect/split/hello-write/read loop for hum_thrum::connect / send_json / read_tones, keeping the nest-specific dispatch (cells, MCP bridge, handle_prompt) in the closure. The cancel / tool-result arms now spawn per-tone to fit the sync read_tones closure, matching the forager's all-spawn pattern. Behavior preserved: hello envelope, MCP bridge, per-sid cell routing, grace-window reconnect policy all unchanged. --- hives/common/src/serve.rs | 67 ++++++++++++++++++--------------------- 1 file changed, 30 insertions(+), 37 deletions(-) diff --git a/hives/common/src/serve.rs b/hives/common/src/serve.rs index 7f4939d..6b46fa2 100644 --- a/hives/common/src/serve.rs +++ b/hives/common/src/serve.rs @@ -27,12 +27,10 @@ use lru::LruCache; use anyhow::{Context, Result}; use serde_json::{json, Value}; -use tokio::io::{AsyncBufReadExt, AsyncWriteExt, BufReader}; -use tokio::net::UnixStream; use tokio::sync::Mutex; use tracing::{debug, info, trace, warn}; -use ensemble::HidPrefix; +use hum_identity::HidPrefix; use mcp::protocol::ToolDef; use nest::{encode_cancel, encode_prompt, encode_tool_result, Cell, Egg, WorkerBee}; use tokio::sync::mpsc; @@ -105,10 +103,7 @@ async fn dial_and_serve( advert: &HiveAdvert, ) -> Result<()> { info!(socket = %path.display(), hive = %advert.hive, "worker.connecting"); - let stream = UnixStream::connect(path).await - .with_context(|| format!("connect to thrum at {}", path.display()))?; - let (read_half, write_half) = stream.into_split(); - let write_half = Arc::new(Mutex::new(write_half)); + let (reader, write_half) = hum_thrum::connect(path).await?; // Load (or mint) the persistent worker-bee identity. The wbee_ // hid survives reconnect + restart; humd indexes manifests by it @@ -137,7 +132,7 @@ async fn dial_and_serve( "chis": ["hello", "prompt", "cancel", "tool-result", "chunk", "finish", "error", "tool-call"], "source": advert.source.clone().unwrap_or_default(), }); - write_half.lock().await.write_all(format!("{}\n", hello).as_bytes()).await?; + hum_thrum::send_json(&write_half, &hello).await?; info!(hive = %advert.hive, hid = %bee_key.hid.short(), models = ?advert.models, "worker.hello.sent"); // Worker-local MCP bridge. The compute (e.g. claude) dials it @@ -148,8 +143,7 @@ async fn dial_and_serve( let bridge = McpBridge::new(Arc::new(move |tone: Value| { let write_half = write_for_bridge.clone(); tokio::spawn(async move { - let line = format!("{}\n", tone); - if let Err(e) = write_half.lock().await.write_all(line.as_bytes()).await { + if let Err(e) = hum_thrum::send_json(&write_half, &tone).await { warn!(err = %e, "mcp.bridge.tool-call.write.failed"); } }); @@ -164,13 +158,7 @@ async fn dial_and_serve( let cells: Arc>> = Arc::new(Mutex::new(LruCache::new(NonZeroUsize::new(MAX_CELLS).unwrap()))); - let mut reader = BufReader::new(read_half).lines(); - while let Some(line) = reader.next_line().await? { - if line.is_empty() { continue; } - let tone: Value = match serde_json::from_str(&line) { - Ok(v) => v, - Err(e) => { trace!(err = %e, "worker.parse.skip"); continue; } - }; + hum_thrum::read_tones(reader, |tone| { let chi = tone.get("chi").and_then(Value::as_str).unwrap_or(""); let sid = tone.get("sid").and_then(Value::as_str).map(str::to_string).unwrap_or_default(); @@ -206,13 +194,17 @@ async fn dial_and_serve( } "cancel" => { if !sid.is_empty() { - let mut r = cells.lock().await; - if let Some(bundle) = r.get(&sid) { - if let Some(rid) = tone.get("rid").and_then(Value::as_str) { - let _ = bundle.stdin.send(encode_cancel(rid)).await; + let cells = cells.clone(); + let tone = tone.clone(); + tokio::spawn(async move { + let mut r = cells.lock().await; + if let Some(bundle) = r.get(&sid) { + if let Some(rid) = tone.get("rid").and_then(Value::as_str) { + let _ = bundle.stdin.send(encode_cancel(rid)).await; + } + bundle.cancel.cancel(); } - bundle.cancel.cancel(); - } + }); } } "tool-result" => { @@ -226,15 +218,19 @@ async fn dial_and_serve( .map(|cid| bridge.resolve(cid, tone.clone())) .unwrap_or(false); if !resolved_by_bridge && !sid.is_empty() { - let mut r = cells.lock().await; - if let Some(bundle) = r.get(&sid) { - if let (Some(call_id), Some(result)) = ( - tone.get("callId").and_then(Value::as_str), - tone.get("result").and_then(Value::as_str), - ) { - let _ = bundle.stdin.send(encode_tool_result(call_id, result)).await; + let cells = cells.clone(); + let tone = tone.clone(); + tokio::spawn(async move { + let mut r = cells.lock().await; + if let Some(bundle) = r.get(&sid) { + if let (Some(call_id), Some(result)) = ( + tone.get("callId").and_then(Value::as_str), + tone.get("result").and_then(Value::as_str), + ) { + let _ = bundle.stdin.send(encode_tool_result(call_id, result)).await; + } } - } + }); } } "breath" | "echo" | "" => { @@ -244,8 +240,7 @@ async fn dial_and_serve( trace!(chi = other, "worker.unknown.chi"); } } - } - Ok(()) + }).await } struct CellBundle { @@ -494,8 +489,7 @@ async fn handle_prompt( "finishReason": if exit_code == 0 { "stop" } else { "error" }, "exitCode": exit_code, }); - let line = format!("{}\n", finish); - let _ = write_for_cleanup.lock().await.write_all(line.as_bytes()).await; + let _ = hum_thrum::send_json(&write_for_cleanup, &finish).await; } cells_for_cleanup.lock().await.pop(&sid_for_cleanup); trace!(sid = %sid_for_cleanup, exit_code, "worker.cell.exit"); @@ -532,8 +526,7 @@ struct WireListener { impl WireListener { async fn send(&self, tone: Value) { - let line = format!("{}\n", tone); - let _ = self.write_half.lock().await.write_all(line.as_bytes()).await; + let _ = hum_thrum::send_json(&self.write_half, &tone).await; } async fn forward_raw(&self, value: Value) { From b5b6d5e74aab0209448d764c5687d787fc1d43f1 Mon Sep 17 00:00:00 2001 From: Adil Date: Mon, 31 Aug 2026 00:47:43 +0500 Subject: [PATCH 5/7] feat(hum-mcp): carve the worker MCP HTTP bridge into a leaf crate MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Move the axum JSON-RPC bridge (spawn_local_mcp / McpBridge / handle) from nest-common into a standalone hum-mcp crate so remote worker hives can expose tools to their compute over MCP without pulling the daemon tree. - New crate hum-mcp: mcp (protocol/catalogue/translate) + thrum-core (rid) + axum + tokio + parking_lot. Keeps the mcp crate pure (no server) — the bridge is the axum layer on top. - hives/common/src/mcp_bridge.rs becomes a thin re-export shim for back-compat; existing nest_common::{spawn_local_mcp, McpBridge} call sites keep resolving. - Moved the bridge tests (tools/list + tools/call round-trip) along. Registered in workspace members; added to hives/common deps. --- Cargo.lock | 16 ++ Cargo.toml | 1 + hives/common/Cargo.toml | 1 + hives/common/src/mcp_bridge.rs | 316 +------------------------------- hum-mcp/Cargo.toml | 20 +++ hum-mcp/src/lib.rs | 319 +++++++++++++++++++++++++++++++++ 6 files changed, 366 insertions(+), 307 deletions(-) create mode 100644 hum-mcp/Cargo.toml create mode 100644 hum-mcp/src/lib.rs diff --git a/Cargo.lock b/Cargo.lock index f561170..d3d6799 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -1928,6 +1928,21 @@ dependencies = [ "tracing", ] +[[package]] +name = "hum-mcp" +version = "0.32.0" +dependencies = [ + "anyhow", + "axum", + "mcp", + "parking_lot", + "serde", + "serde_json", + "thrum-core", + "tokio", + "tracing", +] + [[package]] name = "hum-paths" version = "0.32.0" @@ -3016,6 +3031,7 @@ dependencies = [ "ensemble", "futures", "hum-identity", + "hum-mcp", "hum-paths", "hum-thrum", "ids", diff --git a/Cargo.toml b/Cargo.toml index 2e0955d..b07badd 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -6,6 +6,7 @@ members = [ "hum-paths", "hum-identity", "hum-thrum", + "hum-mcp", "ids", "config", "codegen", diff --git a/hives/common/Cargo.toml b/hives/common/Cargo.toml index fd7cded..5f17252 100644 --- a/hives/common/Cargo.toml +++ b/hives/common/Cargo.toml @@ -9,6 +9,7 @@ description = "Shared building blocks for nests — regex Classifier for the dro hum-paths = { path = "../../hum-paths" } hum-identity = { path = "../../hum-identity" } hum-thrum = { path = "../../hum-thrum" } +hum-mcp = { path = "../../hum-mcp" } drone = { path = "../../drone" } ensemble = { path = "../../ensemble" } mcp = { path = "../../mcp" } diff --git a/hives/common/src/mcp_bridge.rs b/hives/common/src/mcp_bridge.rs index e69f04c..e4db95f 100644 --- a/hives/common/src/mcp_bridge.rs +++ b/hives/common/src/mcp_bridge.rs @@ -1,308 +1,10 @@ -//! Worker-side MCP HTTP server. +//! Worker-side MCP bridge — now a leaf crate. //! -//! Each worker bee that needs to expose tools to its compute via -//! MCP spawns one of these. The bridge serves JSON-RPC at -//! `/s/`, mapping `tools/list` to the worker's current -//! catalogue and `tools/call` to a thrum `chi:"tool-call"` tone -//! through humd. Tool results return as `chi:"tool-result"` tones -//! the worker pumps into the bridge by callId. -//! -//! The mcp/ crate is a pure library — this is where it gets used. - -use std::collections::HashMap; -use std::net::SocketAddr; -use std::sync::Arc; -use std::time::Duration; - -use anyhow::Result; -use axum::extract::{Path, State}; -use axum::http::StatusCode; -use axum::routing::post; -use axum::{Json, Router}; -use parking_lot::{Mutex, RwLock}; -use serde_json::Value; -use tokio::sync::oneshot; -use tracing::{trace, warn}; - -use mcp::protocol::{JsonRpcRequest, JsonRpcResponse, ToolDef}; -use mcp::catalogue; -use mcp::translate; - -/// Shared state between the MCP HTTP handlers and the worker's -/// thrum loop. The worker updates the catalogue on each -/// chi:"prompt" arrival and resolves pending tool-calls when -/// chi:"tool-result" lands. -pub struct McpBridge { - catalogue: RwLock, - pending: Mutex>>, - /// Callback the bridge invokes to ship a `chi:"tool-call"` tone - /// out via the worker's thrum write half. Keeps the bridge - /// transport-agnostic — caller decides how the tone reaches - /// humd. - ship_tool_call: Arc, -} - -#[derive(Debug, Clone, Default)] -struct CatalogueSlot { - tools: Vec, -} - -impl McpBridge { - pub fn new(ship_tool_call: Arc) -> Arc { - Arc::new(Self { - catalogue: RwLock::new(CatalogueSlot::default()), - pending: Mutex::new(HashMap::new()), - ship_tool_call, - }) - } - - /// Set the catalogue for an incoming session. The worker calls - /// this when it receives a chi:"prompt" — both the forager - /// catalogue (humd-merged) and the asker's nestler tools are - /// composed here. `provided` is the capability list (so the - /// merge can filter capability-overlapping nestler tools). - pub fn set_catalogue( - &self, - forager_tools: Vec, - nestler_tools: Vec, - provided: &[String], - ) { - let merged = catalogue::merge(forager_tools, nestler_tools, provided); - *self.catalogue.write() = CatalogueSlot { tools: merged }; - } - - /// Resolve a pending `tools/call` with the result from a - /// `chi:"tool-result"` tone the worker received over thrum. - /// Returns `true` if the callId matched a waiting handler. - pub fn resolve(&self, call_id: &str, tone: Value) -> bool { - if let Some(tx) = self.pending.lock().remove(call_id) { - let _ = tx.send(tone); - true - } else { - false - } - } -} - -/// Spawn the MCP HTTP listener on an ephemeral local port. Returns -/// the bound socket address so the worker can pass it to its -/// compute (e.g. `claude --mcp-config <...url>`). -/// -/// The server runs until the process exits. Spawning blocks only -/// long enough to bind; the listener task is detached. -pub async fn spawn_local_mcp(bridge: Arc) -> Result { - let router = Router::new() - .route("/s/:sid", post(handle)) - .with_state(bridge); - let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await?; - let addr = listener.local_addr()?; - tokio::spawn(async move { - if let Err(e) = axum::serve(listener, router).await { - warn!(err = %e, "mcp.bridge.exit"); - } - }); - Ok(addr) -} - -#[cfg(test)] -mod tests { - use super::*; - use parking_lot::Mutex as PlMutex; - use serde_json::json; - - fn def(name: &str) -> ToolDef { - ToolDef { name: name.into(), description: String::new(), input_schema: json!({}) } - } - - #[tokio::test] - async fn list_tools_returns_set_catalogue() { - let shipped = Arc::new(PlMutex::new(Vec::::new())); - let shipped_for_closure = shipped.clone(); - let bridge = McpBridge::new(Arc::new(move |t| shipped_for_closure.lock().push(t))); - bridge.set_catalogue(vec![def("humfs_read")], vec![], &["fs".into()]); - let addr = spawn_local_mcp(bridge).await.expect("bind"); - let client = reqwest_get_post(); - let body = json!({"jsonrpc":"2.0","id":1,"method":"tools/list","params":{}}); - let resp = client.post(format!("http://{}/s/hum-test", addr)) - .json(&body).send().await.expect("post"); - let v: Value = resp.json().await.expect("json"); - let names: Vec<&str> = v["result"]["tools"].as_array().unwrap() - .iter().map(|t| t["name"].as_str().unwrap()).collect(); - assert!(names.contains(&"humfs_read")); - } - - #[tokio::test] - async fn call_tool_round_trips_via_bridge_resolve() { - let shipped = Arc::new(PlMutex::new(Vec::::new())); - let shipped_for_closure = shipped.clone(); - let bridge = McpBridge::new(Arc::new(move |t| shipped_for_closure.lock().push(t))); - bridge.set_catalogue(vec![def("humfs_read")], vec![], &["fs".into()]); - let addr = spawn_local_mcp(bridge.clone()).await.expect("bind"); - let client = reqwest_get_post(); - // Fire-and-park: post tools/call in a task; after a moment, - // resolve via the bridge with a fake tool-result tone. - let url = format!("http://{}/s/hum-test", addr); - let body = json!({"jsonrpc":"2.0","id":1,"method":"tools/call", - "params":{"name":"humfs_read","arguments":{"file_path":"/x"}}}); - let call_task = tokio::spawn(async move { - client.post(url).json(&body).send().await.unwrap().json().await.unwrap() - }); - // Wait for the bridge to ship the tool-call so we know its callId. - let call_id = tokio::time::timeout(Duration::from_secs(2), async { - loop { - if let Some(tone) = shipped.lock().first().cloned() { - return tone["callId"].as_str().unwrap().to_string(); - } - tokio::time::sleep(Duration::from_millis(10)).await; - } - }).await.expect("tool-call shipped"); - let fake_result = json!({ - "chi":"tool-result","sid":"hum-test","callId":call_id, - "output":"file body" - }); - assert!(bridge.resolve(&call_id, fake_result), "callId resolved"); - let resp = call_task.await.expect("call task joined"); - assert_eq!(resp["result"]["content"][0]["text"], "file body"); - } - - // Minimal reqwest client built from std + hyper would bloat the - // crate; just use a tiny inline TcpStream-based POST helper. - fn reqwest_get_post() -> reqwest_lite::Client { reqwest_lite::Client::new() } - - mod reqwest_lite { - use serde::Serialize; - use serde_json::Value; - use std::io::{Read, Write}; - use std::net::TcpStream; - - pub(crate) struct Client; - impl Client { - pub(crate) fn new() -> Self { Self } - pub(crate) fn post(self, url: String) -> RequestBuilder { - RequestBuilder { url, body: None } - } - } - pub(crate) struct RequestBuilder { - url: String, - body: Option, - } - impl RequestBuilder { - pub(crate) fn json(mut self, v: &T) -> Self { - self.body = Some(serde_json::to_string(v).unwrap()); - self - } - pub(crate) async fn send(self) -> Result { - let url = self.url; - let body = self.body.unwrap_or_default(); - tokio::task::spawn_blocking(move || -> Result { - let stripped = url.strip_prefix("http://").unwrap(); - let (host, path) = stripped.split_once('/').unwrap(); - let path = format!("/{path}"); - let mut stream = TcpStream::connect(host)?; - let req = format!( - "POST {path} HTTP/1.1\r\nHost: {host}\r\nContent-Type: application/json\r\nContent-Length: {}\r\nConnection: close\r\n\r\n{body}", - body.len() - ); - stream.write_all(req.as_bytes())?; - let mut buf = Vec::new(); - stream.read_to_end(&mut buf)?; - let s = String::from_utf8_lossy(&buf).to_string(); - let body_start = s.find("\r\n\r\n").map(|i| i + 4).unwrap_or(0); - Ok(Response { body: s[body_start..].to_string() }) - }).await.unwrap() - } - } - pub(crate) struct Response { body: String } - impl Response { - pub(crate) async fn json(self) -> Result { - serde_json::from_str(&self.body) - } - } - } -} - -async fn handle( - State(bridge): State>, - Path(sid): Path, - Json(req): Json, -) -> (StatusCode, Json>) { - let id = req.id.clone(); - match req.method.as_str() { - "initialize" => (StatusCode::OK, Json(Some(JsonRpcResponse::ok(id, serde_json::json!({ - "protocolVersion": "2024-11-05", - "capabilities": { "tools": {} }, - "serverInfo": { "name": "hum-worker-mcp", "version": "0.30.0" }, - }))))), - - "notifications/initialized" => (StatusCode::OK, Json(None)), - - "tools/list" => { - let slot = bridge.catalogue.read().clone(); - let tools_value = serde_json::to_value(&slot.tools).unwrap_or(serde_json::json!([])); - (StatusCode::OK, Json(Some(JsonRpcResponse::ok( - id, - serde_json::json!({ "tools": tools_value }), - )))) - } - - "tools/call" => { - let params = req.params.unwrap_or(Value::Null); - let name = params.get("name").and_then(Value::as_str).unwrap_or("").to_string(); - if name.is_empty() { - return (StatusCode::OK, Json(Some(JsonRpcResponse::err( - id, -32602, "Missing tool name", - )))); - } - let arguments = params.get("arguments").cloned().unwrap_or(serde_json::json!({})); - let call_id = thrum_core::rid(); - let (tx, rx) = oneshot::channel::(); - bridge.pending.lock().insert(call_id.clone(), tx); - let tone = translate::mcp_call_to_tone(&sid, &call_id, ¶ms); - (bridge.ship_tool_call)(tone); - trace!(%sid, %name, %call_id, "mcp.bridge.tool-call.shipped"); - match tokio::time::timeout(Duration::from_secs(300), rx).await { - Ok(Ok(tone)) => { - let body = translate::tone_to_mcp_result(&tone); - // Mirror the resolution to humd as a sid-tagged - // chi:"chunk" so bee shims (openai-server's - // /v1/responses, anthropic-server's server_tool_use - // path) can surface this tool call as - // provider-executed to the asker. Without this - // mirror, openai-server only sees text + finish - // and OC's openai-responses parser has no way to - // emit a `mcp_call` item with providerExecuted. - let output_text = body.get("content") - .and_then(Value::as_array) - .and_then(|a| a.first()) - .and_then(|p| p.get("text")) - .and_then(Value::as_str) - .unwrap_or("") - .to_string(); - let is_error = body.get("isError").and_then(Value::as_bool).unwrap_or(false); - let chunk = serde_json::json!({ - "chi": "chunk", - "sid": sid, - "chunkType": "tool_executed", - "callId": call_id, - "toolName": name, - "arguments": arguments, - "output": output_text, - "isError": is_error, - }); - (bridge.ship_tool_call)(chunk); - (StatusCode::OK, Json(Some(JsonRpcResponse::ok(id, body)))) - } - _ => { - bridge.pending.lock().remove(&call_id); - (StatusCode::OK, Json(Some(JsonRpcResponse::err( - id, -32000, format!("tool-result for callId {call_id} timed out"), - )))) - } - } - } - - other => (StatusCode::OK, Json(Some(JsonRpcResponse::err( - id, -32601, format!("unknown method '{other}'"), - )))), - } -} +//! The axum JSON-RPC bridge (spawn_local_mcp / McpBridge / handle) +//! lives in the standalone `hum-mcp` crate, so remote worker hives can +//! spawn an MCP server without pulling the daemon tree. This module is +//! a thin re-export shim for back-compat: existing +//! `nest_common::spawn_local_mcp` / `nest_common::McpBridge` call +//! sites keep resolving. + +pub use hum_mcp::{spawn_local_mcp, McpBridge}; diff --git a/hum-mcp/Cargo.toml b/hum-mcp/Cargo.toml new file mode 100644 index 0000000..3e4b1e9 --- /dev/null +++ b/hum-mcp/Cargo.toml @@ -0,0 +1,20 @@ +[package] +name = "hum-mcp" +version.workspace = true +edition.workspace = true +license.workspace = true +description = "Worker-side MCP HTTP bridge. Spawns an axum JSON-RPC server mapping tools/list + tools/call to thrum tones. Leaf crate for remote worker hives — no daemon or nest deps." + +[dependencies] +mcp = { path = "../mcp" } +thrum-core = { path = "../thrum-core" } +serde_json = { workspace = true } +anyhow = { workspace = true } +tracing = { workspace = true } +axum = "0.7" +parking_lot = { workspace = true } +tokio = { workspace = true, features = ["net", "io-util", "sync", "macros", "rt", "time"] } + +[dev-dependencies] +serde = { workspace = true } +tokio = { workspace = true, features = ["macros", "rt"] } diff --git a/hum-mcp/src/lib.rs b/hum-mcp/src/lib.rs new file mode 100644 index 0000000..6d028b0 --- /dev/null +++ b/hum-mcp/src/lib.rs @@ -0,0 +1,319 @@ +//! `hum-mcp` — worker-side MCP HTTP bridge. +//! +//! Each worker bee that needs to expose tools to its compute via MCP +//! spawns one of these. The bridge serves JSON-RPC at +//! `/s/`, mapping `tools/list` to the worker's current +//! catalogue and `tools/call` to a thrum `chi:"tool-call"` tone +//! through humd. Tool results return as `chi:"tool-result"` tones +//! the worker pumps into the bridge by callId. +//! +//! The `mcp` crate is a pure library — this is the axum server that +//! puts it to use. Standalone crate so a remote worker hive can +//! spawn an MCP bridge without pulling the daemon tree. +//! +//! Each worker bee that needs to expose tools to its compute via +//! MCP spawns one of these. The bridge serves JSON-RPC at +//! `/s/`, mapping `tools/list` to the worker's current +//! catalogue and `tools/call` to a thrum `chi:"tool-call"` tone +//! through humd. Tool results return as `chi:"tool-result"` tones +//! the worker pumps into the bridge by callId. +//! +//! The mcp/ crate is a pure library — this is where it gets used. + +use std::collections::HashMap; +use std::net::SocketAddr; +use std::sync::Arc; +use std::time::Duration; + +use anyhow::Result; +use axum::extract::{Path, State}; +use axum::http::StatusCode; +use axum::routing::post; +use axum::{Json, Router}; +use parking_lot::{Mutex, RwLock}; +use serde_json::Value; +use tokio::sync::oneshot; +use tracing::{trace, warn}; + +use mcp::protocol::{JsonRpcRequest, JsonRpcResponse, ToolDef}; +use mcp::catalogue; +use mcp::translate; + +/// Shared state between the MCP HTTP handlers and the worker's +/// thrum loop. The worker updates the catalogue on each +/// chi:"prompt" arrival and resolves pending tool-calls when +/// chi:"tool-result" lands. +pub struct McpBridge { + catalogue: RwLock, + pending: Mutex>>, + /// Callback the bridge invokes to ship a `chi:"tool-call"` tone + /// out via the worker's thrum write half. Keeps the bridge + /// transport-agnostic — caller decides how the tone reaches + /// humd. + ship_tool_call: Arc, +} + +#[derive(Debug, Clone, Default)] +struct CatalogueSlot { + tools: Vec, +} + +impl McpBridge { + pub fn new(ship_tool_call: Arc) -> Arc { + Arc::new(Self { + catalogue: RwLock::new(CatalogueSlot::default()), + pending: Mutex::new(HashMap::new()), + ship_tool_call, + }) + } + + /// Set the catalogue for an incoming session. The worker calls + /// this when it receives a chi:"prompt" — both the forager + /// catalogue (humd-merged) and the asker's nestler tools are + /// composed here. `provided` is the capability list (so the + /// merge can filter capability-overlapping nestler tools). + pub fn set_catalogue( + &self, + forager_tools: Vec, + nestler_tools: Vec, + provided: &[String], + ) { + let merged = catalogue::merge(forager_tools, nestler_tools, provided); + *self.catalogue.write() = CatalogueSlot { tools: merged }; + } + + /// Resolve a pending `tools/call` with the result from a + /// `chi:"tool-result"` tone the worker received over thrum. + /// Returns `true` if the callId matched a waiting handler. + pub fn resolve(&self, call_id: &str, tone: Value) -> bool { + if let Some(tx) = self.pending.lock().remove(call_id) { + let _ = tx.send(tone); + true + } else { + false + } + } +} + +/// Spawn the MCP HTTP listener on an ephemeral local port. Returns +/// the bound socket address so the worker can pass it to its +/// compute (e.g. `claude --mcp-config <...url>`). +/// +/// The server runs until the process exits. Spawning blocks only +/// long enough to bind; the listener task is detached. +pub async fn spawn_local_mcp(bridge: Arc) -> Result { + let router = Router::new() + .route("/s/:sid", post(handle)) + .with_state(bridge); + let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await?; + let addr = listener.local_addr()?; + tokio::spawn(async move { + if let Err(e) = axum::serve(listener, router).await { + warn!(err = %e, "mcp.bridge.exit"); + } + }); + Ok(addr) +} + +#[cfg(test)] +mod tests { + use super::*; + use parking_lot::Mutex as PlMutex; + use serde_json::json; + + fn def(name: &str) -> ToolDef { + ToolDef { name: name.into(), description: String::new(), input_schema: json!({}) } + } + + #[tokio::test] + async fn list_tools_returns_set_catalogue() { + let shipped = Arc::new(PlMutex::new(Vec::::new())); + let shipped_for_closure = shipped.clone(); + let bridge = McpBridge::new(Arc::new(move |t| shipped_for_closure.lock().push(t))); + bridge.set_catalogue(vec![def("humfs_read")], vec![], &["fs".into()]); + let addr = spawn_local_mcp(bridge).await.expect("bind"); + let client = reqwest_get_post(); + let body = json!({"jsonrpc":"2.0","id":1,"method":"tools/list","params":{}}); + let resp = client.post(format!("http://{}/s/hum-test", addr)) + .json(&body).send().await.expect("post"); + let v: Value = resp.json().await.expect("json"); + let names: Vec<&str> = v["result"]["tools"].as_array().unwrap() + .iter().map(|t| t["name"].as_str().unwrap()).collect(); + assert!(names.contains(&"humfs_read")); + } + + #[tokio::test] + async fn call_tool_round_trips_via_bridge_resolve() { + let shipped = Arc::new(PlMutex::new(Vec::::new())); + let shipped_for_closure = shipped.clone(); + let bridge = McpBridge::new(Arc::new(move |t| shipped_for_closure.lock().push(t))); + bridge.set_catalogue(vec![def("humfs_read")], vec![], &["fs".into()]); + let addr = spawn_local_mcp(bridge.clone()).await.expect("bind"); + let client = reqwest_get_post(); + // Fire-and-park: post tools/call in a task; after a moment, + // resolve via the bridge with a fake tool-result tone. + let url = format!("http://{}/s/hum-test", addr); + let body = json!({"jsonrpc":"2.0","id":1,"method":"tools/call", + "params":{"name":"humfs_read","arguments":{"file_path":"/x"}}}); + let call_task = tokio::spawn(async move { + client.post(url).json(&body).send().await.unwrap().json().await.unwrap() + }); + // Wait for the bridge to ship the tool-call so we know its callId. + let call_id = tokio::time::timeout(Duration::from_secs(2), async { + loop { + if let Some(tone) = shipped.lock().first().cloned() { + return tone["callId"].as_str().unwrap().to_string(); + } + tokio::time::sleep(Duration::from_millis(10)).await; + } + }).await.expect("tool-call shipped"); + let fake_result = json!({ + "chi":"tool-result","sid":"hum-test","callId":call_id, + "output":"file body" + }); + assert!(bridge.resolve(&call_id, fake_result), "callId resolved"); + let resp = call_task.await.expect("call task joined"); + assert_eq!(resp["result"]["content"][0]["text"], "file body"); + } + + // Minimal reqwest client built from std + hyper would bloat the + // crate; just use a tiny inline TcpStream-based POST helper. + fn reqwest_get_post() -> reqwest_lite::Client { reqwest_lite::Client::new() } + + mod reqwest_lite { + use serde::Serialize; + use serde_json::Value; + use std::io::{Read, Write}; + use std::net::TcpStream; + + pub(crate) struct Client; + impl Client { + pub(crate) fn new() -> Self { Self } + pub(crate) fn post(self, url: String) -> RequestBuilder { + RequestBuilder { url, body: None } + } + } + pub(crate) struct RequestBuilder { + url: String, + body: Option, + } + impl RequestBuilder { + pub(crate) fn json(mut self, v: &T) -> Self { + self.body = Some(serde_json::to_string(v).unwrap()); + self + } + pub(crate) async fn send(self) -> Result { + let url = self.url; + let body = self.body.unwrap_or_default(); + tokio::task::spawn_blocking(move || -> Result { + let stripped = url.strip_prefix("http://").unwrap(); + let (host, path) = stripped.split_once('/').unwrap(); + let path = format!("/{path}"); + let mut stream = TcpStream::connect(host)?; + let req = format!( + "POST {path} HTTP/1.1\r\nHost: {host}\r\nContent-Type: application/json\r\nContent-Length: {}\r\nConnection: close\r\n\r\n{body}", + body.len() + ); + stream.write_all(req.as_bytes())?; + let mut buf = Vec::new(); + stream.read_to_end(&mut buf)?; + let s = String::from_utf8_lossy(&buf).to_string(); + let body_start = s.find("\r\n\r\n").map(|i| i + 4).unwrap_or(0); + Ok(Response { body: s[body_start..].to_string() }) + }).await.unwrap() + } + } + pub(crate) struct Response { body: String } + impl Response { + pub(crate) async fn json(self) -> Result { + serde_json::from_str(&self.body) + } + } + } +} + +async fn handle( + State(bridge): State>, + Path(sid): Path, + Json(req): Json, +) -> (StatusCode, Json>) { + let id = req.id.clone(); + match req.method.as_str() { + "initialize" => (StatusCode::OK, Json(Some(JsonRpcResponse::ok(id, serde_json::json!({ + "protocolVersion": "2024-11-05", + "capabilities": { "tools": {} }, + "serverInfo": { "name": "hum-worker-mcp", "version": "0.30.0" }, + }))))), + + "notifications/initialized" => (StatusCode::OK, Json(None)), + + "tools/list" => { + let slot = bridge.catalogue.read().clone(); + let tools_value = serde_json::to_value(&slot.tools).unwrap_or(serde_json::json!([])); + (StatusCode::OK, Json(Some(JsonRpcResponse::ok( + id, + serde_json::json!({ "tools": tools_value }), + )))) + } + + "tools/call" => { + let params = req.params.unwrap_or(Value::Null); + let name = params.get("name").and_then(Value::as_str).unwrap_or("").to_string(); + if name.is_empty() { + return (StatusCode::OK, Json(Some(JsonRpcResponse::err( + id, -32602, "Missing tool name", + )))); + } + let arguments = params.get("arguments").cloned().unwrap_or(serde_json::json!({})); + let call_id = thrum_core::rid(); + let (tx, rx) = oneshot::channel::(); + bridge.pending.lock().insert(call_id.clone(), tx); + let tone = translate::mcp_call_to_tone(&sid, &call_id, ¶ms); + (bridge.ship_tool_call)(tone); + trace!(%sid, %name, %call_id, "mcp.bridge.tool-call.shipped"); + match tokio::time::timeout(Duration::from_secs(300), rx).await { + Ok(Ok(tone)) => { + let body = translate::tone_to_mcp_result(&tone); + // Mirror the resolution to humd as a sid-tagged + // chi:"chunk" so bee shims (openai-server's + // /v1/responses, anthropic-server's server_tool_use + // path) can surface this tool call as + // provider-executed to the asker. Without this + // mirror, openai-server only sees text + finish + // and OC's openai-responses parser has no way to + // emit a `mcp_call` item with providerExecuted. + let output_text = body.get("content") + .and_then(Value::as_array) + .and_then(|a| a.first()) + .and_then(|p| p.get("text")) + .and_then(Value::as_str) + .unwrap_or("") + .to_string(); + let is_error = body.get("isError").and_then(Value::as_bool).unwrap_or(false); + let chunk = serde_json::json!({ + "chi": "chunk", + "sid": sid, + "chunkType": "tool_executed", + "callId": call_id, + "toolName": name, + "arguments": arguments, + "output": output_text, + "isError": is_error, + }); + (bridge.ship_tool_call)(chunk); + (StatusCode::OK, Json(Some(JsonRpcResponse::ok(id, body)))) + } + _ => { + bridge.pending.lock().remove(&call_id); + (StatusCode::OK, Json(Some(JsonRpcResponse::err( + id, -32000, format!("tool-result for callId {call_id} timed out"), + )))) + } + } + } + + other => (StatusCode::OK, Json(Some(JsonRpcResponse::err( + id, -32601, format!("unknown method '{other}'"), + )))), + } +} From 5cfc393b79cc70491ad23ca4cd83c990b1effa05 Mon Sep 17 00:00:00 2001 From: Adil Date: Mon, 31 Aug 2026 00:52:22 +0500 Subject: [PATCH 6/7] refactor(tooldef): unify ToolDef/ToolResult on mcp::protocol MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit forager.rs defined its own ToolDef/ToolResult with the same shape as mcp::protocol's. Remove the duplicates and use mcp::protocol::{ToolDef, ToolResult} everywhere; nest-common now re-exports them from mcp. humfs and the forager dispatch construct both types identically, so the unify is a no-op at call sites (mcp's ToolResult also derives Serialize/Default + serde attrs). No separate hum-tooldef crate — mcp is already a pure library, so the shared tool-surface types re-export straight from it (per the doc's '...or just from mcp' option). --- hives/common/src/forager.rs | 40 +------------------------------------ hives/common/src/lib.rs | 3 ++- 2 files changed, 3 insertions(+), 40 deletions(-) diff --git a/hives/common/src/forager.rs b/hives/common/src/forager.rs index eb471ee..3c0ed50 100644 --- a/hives/common/src/forager.rs +++ b/hives/common/src/forager.rs @@ -29,50 +29,12 @@ use std::sync::Arc; use anyhow::{Context, Result}; use async_trait::async_trait; use hum_identity::HidPrefix; +use mcp::protocol::{ToolDef, ToolResult}; use serde_json::{json, Value}; use tracing::{info, trace}; use crate::identity::load_or_mint_bee_key; -/// One advertised tool. Description + schema land in humd's tool -/// registry and get fanned out to MCP clients verbatim. -#[derive(Debug, Clone, Default)] -pub struct ToolDef { - /// Tool name — `humfs_read`, `humfs_do_code`, etc. Routing key. - pub name: String, - /// Free-form description; rendered by MCP clients in their tool - /// pickers. - pub description: String, - /// JSON schema for the tool's `args` object. Foragers MUST - /// validate `args` against this themselves before dispatching — - /// humd does not enforce schemas. - pub input_schema: Value, -} - -/// Outcome of one tool dispatch. -#[derive(Debug, Clone)] -pub struct ToolResult { - /// Free-form text rendered to the asker. Tool authors decide - /// shape (e.g. line-numbered file slice, hit list, status line). - pub output: String, - /// Optional short title shown in the asker's tool-call header. - pub title: Option, - /// Optional structured side-channel data (e.g. image base64, - /// usage stats). - pub metadata: Option, - /// True if dispatch failed; output carries the error message. - pub is_error: bool, -} - -impl ToolResult { - pub fn text(s: impl Into) -> Self { - Self { output: s.into(), title: None, metadata: None, is_error: false } - } - pub fn error(s: impl Into) -> Self { - Self { output: s.into(), title: None, metadata: None, is_error: true } - } -} - /// Forager-side tool dispatcher. The forager binary owns its own /// state (cwd, fs.roots snapshot, permission cache); this trait is /// the seam humd's tool-call router calls into. diff --git a/hives/common/src/lib.rs b/hives/common/src/lib.rs index 2205457..1286451 100644 --- a/hives/common/src/lib.rs +++ b/hives/common/src/lib.rs @@ -14,7 +14,8 @@ pub mod identity; pub mod mcp_bridge; pub mod serve; pub mod suspicion_regex; -pub use forager::{serve_forager, ForagerAdvert, ToolDef, ToolDispatcher, ToolResult}; +pub use forager::{serve_forager, ForagerAdvert, ToolDispatcher}; +pub use mcp::protocol::{ToolDef, ToolResult}; pub use identity::{bee_key_path, load_or_mint_bee_key, BeeKey}; pub use mcp_bridge::{spawn_local_mcp, McpBridge}; pub use serve::{serve_worker, HiveAdvert}; From a39c878e6f8b9f055d42e4a30cc4b3aedb2b51dd Mon Sep 17 00:00:00 2001 From: Adil Date: Mon, 31 Aug 2026 00:54:13 +0500 Subject: [PATCH 7/7] docs(reusable-crates): mark the carve as landed, list what's done --- hives/common/REUSABLE_CRATES.md | 206 ++++++++++++++++++++++++++++++++ 1 file changed, 206 insertions(+) create mode 100644 hives/common/REUSABLE_CRATES.md diff --git a/hives/common/REUSABLE_CRATES.md b/hives/common/REUSABLE_CRATES.md new file mode 100644 index 0000000..70cd52f --- /dev/null +++ b/hives/common/REUSABLE_CRATES.md @@ -0,0 +1,206 @@ +# Reusable crates for remote hives + +> Status: **implemented on branch `crates/reusable-hive-libs`**. +> Scope: which building blocks a *remote hive* — a hive whose source +> lives in a foreign repo and is shallow-cloned by `hum hive install` — +> should be able to depend on as published crates, instead of reaching +> back into `hum`'s internal tree. +> +> The carve is landed as six commits: the four `hum-*` leaf crates exist, +> `Hid` moved to `ids`, and the tool-surface types unify on `mcp`. The +> daemon tree (`nest`, `drone`, `ensemble`, `mcp`) is unchanged and still +> not imported by remote hives. + +## The goal + +A hive is a kind + a binary. The whole point of the ensemble is that a +bee on one machine reaches a humd on another, and nothing installs on +the remote humd's disk. So a remote hive's *runtime* only needs the +wire: connect, hello, route tones. It does **not** need the daemon's +in-memory nest, drone, or ensemble machinery. + +But today every Rust hive that wants the convenience helpers — identity +minting, `serve_worker` / `serve_forager` loops, MCP bridge, tool-def +shapes — pulls in `nest-common` (`hives/common`), which transitively +drags `nest`, `drone`, `ensemble`, `mcp`, `hum-paths`. That is the +daemon's whole dependency tree for what should be ~four small leaf +crates. A foreign hive shouldn't need `iroh`, `rustls`, `command-group`, +`portable-pty`, `sysinfo`, `metrics` just to mint a bee key and dial a +Unix socket. + +This doc evaluates *what* should be carved out, *where* it should live, +and *what stays put*. The boundary is now landed — see **Done** below. + +## Current coupling (measured) + +`nest-common` (`hives/common`, 1606 lines) is the shared hive library: + +| module | lines | pulls in | used by | +|---|---|---|---| +| `serve.rs` (`serve_worker`) | 749 | nest, mcp, hum-paths, ensemble(HidPrefix), ids, lru, metrics | claude-cli, claude-repl, ollama-worker | +| `forager.rs` (`serve_forager`) | 219 | ensemble(HidPrefix), hum-paths, thrum-core | humfs | +| `identity.rs` (`load_or_mint_bee_key`) | 158 | ed25519-dalek, ensemble(Hid), hum-paths, rand | grpc, gsm-modem, bp7, paid-oracle, + all above | +| `mcp_bridge.rs` | 308 | axum, mcp, thrum-core, tokio | claude-cli, claude-repl, ollama-worker | +| `suspicion_regex.rs` | 151 | drone, regex | drone's context-loss heuristic (not hive-facing today) | + +The heaviest transitive pull is `ensemble` (iroh QUIC, rustls, tokio-rustls, +onchain reqwest) — yet the hives only touch its `Hid`/`HidPrefix` types, a +~40-line pure-crypto block. `nest` (command-group, portable-pty, sysinfo, +metrics, libc) is only needed by `serve_worker`, not by foragers. + +## What should be reusable crates (remote hives) + +These are the *wire-facing* primitives a remote hive needs. Each is a leaf +or near-leaf: no daemon state, no transport, no LLM subprocess machinery. + +### 1. `hum-identity` — bee identity minting (`identity.rs`) + +What remote hives actually need on boot: load-or-mint a 32-byte ed25519 +seed at `$XDG_STATE_HOME/hum/bees/.key`, derive the role-tagged +`fbee_` / `wbee_` Hid. This is the **mandatory** hello field — +without it humd can't dedupe across reconnects and leaks a fresh manifest +per reconnect (see hives/README's hid warning). + +Today it lives in `nest-common::identity` and pulls `ensemble` (for Hid) ++ `hum-paths`. Both are heavier than needed: +- `Hid`/`HidPrefix` is a 40-line pure sha256+hex type that should move to + the `ids` crate (which already owns `HumId`) or a new `hum-hid` leaf. +- `hum_paths::bee_key` is one function; the seed format + path must stay + byte-identical with the TS (`openai-server/src/identity.ts`) and Go + (`twilio-sms/main.go`) implementations, which already hardcode the path. + +**Boundary:** a `hum-identity` crate exposing `BeeKey { signing, hid }`, +`load_or_mint_bee_key(kind, prefix)`, `bee_key_path(kind)`. Deps: +`ed25519-dalek`, `rand`, `ids` (for Hid), `hum-paths` (or inline the path — +see below). No ensemble, no nest, no mcp. + +### 2. `hum-thrum` — the wire client (`serve_forager` core + a worker half) + +The thrum client loop is the single most reusable thing in the repo: dial +the socket, send `chi:"hello"`, read NDJSON tones, dispatch by chi, ship +results, reconnect forever. Today it's fused into `serve_worker` (nest- +specific) and `serve_forager` (tool-dispatcher-specific), so neither is +usable by a hive that isn't exactly a worker or exactly a forager. + +**Boundary:** a transport-agnostic `hum-thrum` crate: +- `connect()` + `hello(advert)` + read/write half split + reconnect loop + (the pattern in both serve.rs and forager.rs, factored out) +- chi dispatch table (`tool-call`, `prompt`, `cancel`, `tool-result`, …) + built from `thrum-core::Chi` +- `rid()` already lives in `thrum-core`. +Deps: `tokio`, `serde_json`, `thrum-core`, `hum-paths` (socket path). +Then `serve_worker` / `serve_forager` become thin adapters over it. + +### 3. `hum-mcp` — worker-side MCP bridge (`mcp_bridge.rs` + `mcp`) + +`mcp` is already a pure library (JSON-RPC envelope, ToolDef, capability +table, tone↔request mapping) — that's the right shape. But the actual +*bridge* (`spawn_local_mcp`, `McpBridge` with pending oneshot resolution) +currently sits in `nest-common::mcp_bridge` with an axum dependency and a +closure callback that ships tones. A worker hive that wants to expose tools +to its compute over MCP needs exactly this. + +**Boundary:** fold `mcp_bridge.rs` into the `mcp` crate as a `bridge` +module (axum server + pending map), so `serve_worker` and any remote +worker hive share it. Deps: `mcp`, `axum`, `tokio`, `thrum-core`. + +### 4. `hum-tooldef` — tool surface shapes (`ToolDef`, `ToolResult`) + +humfs and every forager hive declare tools. Today `ToolDef`/`ToolResult` +are defined in `forager.rs` while `mcp::protocol::ToolDef` is a *different* +type with the same shape. Remote hives need one shared type. + +**Boundary:** unify on `mcp::protocol::ToolDef` + `ToolResult` (already +pure) and re-export from a `hum-tooldef` crate (or just from `mcp`). +humfs already imports `ToolDef` from `nest_common`; point it at `mcp`. + +### 5. `hum-chi` — the chi registry (already exists: `thrum-core`) + +Remote hives need the chi enum, `THRUM_VERSION`, `rid`, `sigil`, envelope, +wane. That's `thrum-core` today — already a leaf (deps: `ids`, sha2, hex, +strum). This is the model for the others: **a remote hive should depend on +`thrum-core` and `hum-identity` and `hum-thrum`, not on `nest-common`.** + +`thrum-core` is also the source of truth that codegen fans out to +TS/Python/Go clients — so it must stay in-repo and stay the canonical enum. + +## What should NOT be reusable crates + +Leave these in the daemon tree; they are not wire-facing and a remote hive +has no business compiling them: + +| thing | why it stays | +|---|---| +| `serve_worker`'s nest machinery (`Cell`, `Egg`, `WorkerBee`, LRU cell pool, idle reaper) | daemon-owned compute lifecycle; the `nest` crate is the LLM subprocess pool. A remote hive that *is* a worker implements `WorkerBee` and uses `nest` — that's the point of a worker — but this is the heavy path, and only workers need it | +| `serve_forager`'s `ToolDispatcher` trait + dispatch loop | the per-tool runtime is hive-specific state (cwd, fs roots, permission cache); the *wire* part (see `hum-thrum`) is what's shared | +| `drone` + `suspicion_regex` | sentinel/classifier, daemon-side observability; not hive-facing | +| `ensemble` itself (iroh/rustls/tcp/tls/gossip/kad) | the mesh is humd-to-humd. A bee reaches a *remote* humd over the ensemble *transport* hosted by that humd; the hive never opens the mesh itself. Its only need is `Hid`, which moves to `ids`/`hum-hid` | +| `hum-paths` full | a remote hive needs two paths (`thrum_sock_resolved`, `bee_key`). Either keep `hum-paths` as a tiny leaf (it already is: serde only) or fold those two into `hum-identity`/`hum-thrum` so a hive doesn't import the whole path module. The TS/Go impls already inline these paths, so inlining in Rust keeps parity | + +## The dependency ladder + +``` +thrum-core (chi, rid, sigil, envelope) ← leaf, canonical +ids (HumId + Hid/HidPrefix after move) ← leaf +hum-paths (socket + bee-key paths) ← leaf, serde only +hum-identity = ids + ed25519-dalek + rand + hum-paths +hum-thrum = thrum-core + tokio + serde_json + hum-paths +hum-tooldef = mcp::protocol (pure) +hum-mcp = mcp + axum + tokio (bridge) + +remote hive → thrum-core, hum-identity, hum-thrum (+ hum-mcp / hum-tooldef as needed) +daemon tree → nest, drone, ensemble, mcp (unchanged, not imported by remote hives) +``` + +The win: a remote Rust hive goes from importing `nest-common` (→ ensemble's +iroh/rustls, nest's command-group/portable-pty/sysinfo, metrics) to importing +three or four small leaf crates whose combined deps are tokio + serde + +ed25519 + sha2 + hex + axum (for MCP). + +## Done (this branch) + +Landed as `c3cd7a6..5cfc393`: + +1. `ids` — `Hid`/`HidPrefix`/`HidParseError` moved out of `ensemble`; + `ensemble` re-exports `ids::{Hid, HidPrefix, HidParseError}` for + back-compat. `ids` gains a `hex` dep. +2. `hum-identity` — `load_or_mint_bee_key` / `BeeKey` / `bee_key_path` + carved into a leaf crate (`ids` + `hum-paths` + `ed25519-dalek` + + `rand`). `hives/common::identity` is a thin re-export shim. +3. `hum-thrum` — the wire client (`connect` / `send_json` / + `read_tones` / `serve_forever`) carved into a leaf crate + (`ids` + `hum-paths` + `hum-identity` + `thrum-core` + `tokio`). + `serve_forager` and `serve_worker` now use it; the chi semantics stay + in `nest-common`. +4. `hum-mcp` — the axum JSON-RPC bridge (`spawn_local_mcp` / + `McpBridge` / `handle`) carved into a leaf crate (`mcp` + + `thrum-core` + `axum` + `tokio`). `hives/common::mcp_bridge` is a thin + re-export shim. The `mcp` crate stays a pure library (no server). +5. Tool surface unified on `mcp::protocol::{ToolDef, ToolResult}` — + the `forager.rs` duplicates are gone; `nest-common` re-exports from + `mcp`. (Chose the doc's "…or just from `mcp`" option — no separate + `hum-tooldef` crate, since `mcp` is already pure.) + +Remaining from the original proposal (not carved yet, still in +`nest-common`): `suspicion_regex.rs` (sentinel heuristic — not hive- +facing), and the `serve_worker` nest machinery (`Cell`/`Egg`/LRU pool — +daemon-owned compute lifecycle, only workers need it). + +## Migration + +1. Move `Hid`/`HidPrefix` from `ensemble/src/lib.rs` into `ids` (or a new + `hum-hid`); re-export from `ensemble` for back-compat so existing + `ensemble::Hid` call sites keep compiling. +2. Carve `identity.rs` → `hum-identity` crate. +3. Factor the thrum client loop out of `serve.rs`/`forager.rs` → `hum-thrum`. +4. Fold `mcp_bridge.rs` into `mcp` as `bridge`. +5. Re-point every hive's `nest-common::{...}` import at the new crates; + keep `nest-common` as a thin re-export shim during the transition so + the non-Rust hives' docs (which reference `nest_common::load_or_mint_bee_key`) + keep resolving. +6. Add each new crate to `[workspace] members` + `[workspace.dependencies]`. + +Each new crate is publishable independently (crates.io or a git source), +which is exactly what a remote hive's `Orchfile`/`source` needs: point the +hive at the published `hum-thrum` + `hum-identity` instead of the whole hum +repo.