From 8cceaffbce72e7aa12ae75e58c3eaeb039c9696a Mon Sep 17 00:00:00 2001 From: Flach Date: Tue, 28 Jul 2026 14:32:40 -0500 Subject: [PATCH] =?UTF-8?q?feat(kannaka):=20buzz-kannaka=20adapter=20crate?= =?UTF-8?q?=20=E2=80=94=20HRM=20memory=20as=20a=20service=20(v0)?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Adds crates/buzz-kannaka, the first Kannaka glue crate from the docs/KANNAKA.md roadmap: an async MemoryService trait (remember / recall / status) with a subprocess-backed KannakaCli implementation following kannaka-memory's ADR-0016 integration contract (CLI binary, JSON stdout, diagnostics on stderr). Also exposes dream, observe --json, and forget. Purely additive — no buzz-core/buzz-relay changes. Co-Authored-By: Claude Fable 5 --- Cargo.lock | 12 ++ Cargo.toml | 1 + crates/buzz-kannaka/Cargo.toml | 16 +++ crates/buzz-kannaka/src/client.rs | 193 ++++++++++++++++++++++++++++++ crates/buzz-kannaka/src/error.rs | 52 ++++++++ crates/buzz-kannaka/src/lib.rs | 58 +++++++++ crates/buzz-kannaka/src/types.rs | 87 ++++++++++++++ docs/KANNAKA.md | 7 +- 8 files changed, 425 insertions(+), 1 deletion(-) create mode 100644 crates/buzz-kannaka/Cargo.toml create mode 100644 crates/buzz-kannaka/src/client.rs create mode 100644 crates/buzz-kannaka/src/error.rs create mode 100644 crates/buzz-kannaka/src/lib.rs create mode 100644 crates/buzz-kannaka/src/types.rs diff --git a/Cargo.lock b/Cargo.lock index 3b60dc4579f..c753c7ab8ff 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -990,6 +990,18 @@ dependencies = [ "zeroize", ] +[[package]] +name = "buzz-kannaka" +version = "0.1.0" +dependencies = [ + "serde", + "serde_json", + "thiserror 2.0.18", + "tokio", + "tracing", + "uuid", +] + [[package]] name = "buzz-media" version = "0.1.0" diff --git a/Cargo.toml b/Cargo.toml index 3ac7ee4cce1..ba7b00871ec 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -26,6 +26,7 @@ members = [ "crates/buzz-pair-relay", "crates/buzz-relay-mesh", "crates/buzz-dev-mcp", + "crates/buzz-kannaka", "examples/countdown-bot", ] exclude = ["desktop/src-tauri"] diff --git a/crates/buzz-kannaka/Cargo.toml b/crates/buzz-kannaka/Cargo.toml new file mode 100644 index 00000000000..5181b551683 --- /dev/null +++ b/crates/buzz-kannaka/Cargo.toml @@ -0,0 +1,16 @@ +[package] +name = "buzz-kannaka" +version.workspace = true +edition.workspace = true +rust-version.workspace = true +license.workspace = true +repository.workspace = true +description = "Kannaka HRM memory adapter — recall/remember/observe/dream for Buzz agents and workflows" + +[dependencies] +tokio = { workspace = true } +serde = { workspace = true } +serde_json = { workspace = true } +thiserror = { workspace = true } +tracing = { workspace = true } +uuid = { workspace = true } diff --git a/crates/buzz-kannaka/src/client.rs b/crates/buzz-kannaka/src/client.rs new file mode 100644 index 00000000000..736601ef495 --- /dev/null +++ b/crates/buzz-kannaka/src/client.rs @@ -0,0 +1,193 @@ +use crate::error::KannakaError; +use crate::types::{RecallResult, RememberOptions, SystemStatus}; +use crate::MemoryService; +use std::path::PathBuf; +use std::process::Stdio; +use std::time::Duration; +use tokio::process::Command; +use uuid::Uuid; + +/// Environment variable overriding the `kannaka` binary path. +pub const BIN_ENV: &str = "BUZZ_KANNAKA_BIN"; +/// Environment variable naming the HRM data directory, passed through to the CLI. +pub const DATA_DIR_ENV: &str = "KANNAKA_DATA_DIR"; + +const DEFAULT_TIMEOUT: Duration = Duration::from_secs(30); + +/// Subprocess-backed Kannaka client (ADR-0016 contract: CLI + JSON stdout). +#[derive(Debug, Clone)] +pub struct KannakaCli { + bin: PathBuf, + data_dir: Option, + timeout: Duration, +} + +impl Default for KannakaCli { + fn default() -> Self { + Self::new() + } +} + +impl KannakaCli { + /// Client using `$BUZZ_KANNAKA_BIN` (or `kannaka` on `PATH`) and the + /// CLI's own data-dir resolution (`$KANNAKA_DATA_DIR` / `~/.kannaka`). + pub fn new() -> Self { + let bin = std::env::var_os(BIN_ENV) + .map(PathBuf::from) + .unwrap_or_else(|| PathBuf::from("kannaka")); + Self { + bin, + data_dir: None, + timeout: DEFAULT_TIMEOUT, + } + } + + /// Pin the HRM data directory (exported as `KANNAKA_DATA_DIR` to the child). + pub fn with_data_dir(mut self, dir: impl Into) -> Self { + self.data_dir = Some(dir.into()); + self + } + + /// Override the per-command deadline (default 30 s). + pub fn with_timeout(mut self, timeout: Duration) -> Self { + self.timeout = timeout; + self + } + + /// Run `kannaka ` and return trimmed stdout. + async fn run(&self, args: &[&str]) -> Result { + let command = args.first().copied().unwrap_or_default().to_string(); + let mut cmd = Command::new(&self.bin); + cmd.args(args) + .stdin(Stdio::null()) + .stdout(Stdio::piped()) + .stderr(Stdio::piped()) + .kill_on_drop(true); + if let Some(dir) = &self.data_dir { + cmd.env(DATA_DIR_ENV, dir); + } + tracing::debug!(bin = %self.bin.display(), ?args, "invoking kannaka CLI"); + + let child = cmd.spawn().map_err(|source| KannakaError::Spawn { + bin: self.bin.display().to_string(), + source, + })?; + let output = tokio::time::timeout(self.timeout, child.wait_with_output()) + .await + .map_err(|_| KannakaError::Timeout { + command: command.clone(), + timeout: self.timeout, + })? + .map_err(|source| KannakaError::Spawn { + bin: self.bin.display().to_string(), + source, + })?; + + if !output.status.success() { + return Err(KannakaError::CommandFailed { + command, + status: output.status, + stderr: String::from_utf8_lossy(&output.stderr).trim().to_string(), + }); + } + let stdout = + String::from_utf8(output.stdout).map_err(|_| KannakaError::NonUtf8 { command })?; + Ok(stdout.trim().to_string()) + } + + /// Run a dream consolidation cycle; returns the CLI's human-readable + /// report verbatim (the CLI does not emit JSON for `dream`). + pub async fn dream(&self, deep: bool) -> Result { + let mode = if deep { "deep" } else { "lite" }; + self.run(&["dream", "--mode", mode]).await + } + + /// Full `observe --json` system report, untyped. + pub async fn observe(&self) -> Result { + let out = self.run(&["observe", "--json"]).await?; + serde_json::from_str(&out).map_err(|source| KannakaError::Parse { + command: "observe".into(), + source, + }) + } + + /// Delete a memory by id. + pub async fn forget(&self, id: Uuid) -> Result<(), KannakaError> { + self.run(&["forget", &id.to_string()]).await.map(|_| ()) + } +} + +impl MemoryService for KannakaCli { + async fn remember(&self, text: &str, opts: RememberOptions) -> Result { + let mut args: Vec = vec!["remember".into(), text.into()]; + if let Some(importance) = opts.importance { + args.extend(["--importance".into(), importance.to_string()]); + } + if let Some(category) = &opts.category { + args.extend(["--category".into(), category.clone()]); + } + if let Some(modality) = &opts.modality { + args.extend(["--modality".into(), modality.clone()]); + } + if !opts.tags.is_empty() { + args.push("--tags".into()); + args.extend(opts.tags.iter().cloned()); + } + let arg_refs: Vec<&str> = args.iter().map(String::as_str).collect(); + let out = self.run(&arg_refs).await?; + // `remember` prints the bare UUID of the new memory. + out.parse().map_err(|_| KannakaError::Parse { + command: "remember".into(), + source: serde_json::Error::io(std::io::Error::new( + std::io::ErrorKind::InvalidData, + format!("expected UUID on stdout, got: {out}"), + )), + }) + } + + async fn recall(&self, query: &str, top_k: usize) -> Result, KannakaError> { + let k = top_k.to_string(); + let out = self.run(&["recall", query, "--top-k", &k]).await?; + if out.is_empty() { + return Ok(Vec::new()); + } + serde_json::from_str(&out).map_err(|source| KannakaError::Parse { + command: "recall".into(), + source, + }) + } + + async fn status(&self) -> Result { + let out = self.run(&["status"]).await?; + serde_json::from_str(&out).map_err(|source| KannakaError::Parse { + command: "status".into(), + source, + }) + } +} + +#[cfg(test)] +mod tests { + use super::*; + + #[tokio::test] + async fn missing_binary_is_a_spawn_error() { + let client = KannakaCli { + bin: PathBuf::from("kannaka-definitely-not-installed"), + data_dir: None, + timeout: DEFAULT_TIMEOUT, + }; + let err = client.recall("anything", 3).await.unwrap_err(); + assert!(matches!(err, KannakaError::Spawn { .. }), "got: {err}"); + } + + /// Round-trip against a real `kannaka` binary when one is on PATH. + /// Ignored by default so CI without the constellation stays green. + #[tokio::test] + #[ignore = "requires a kannaka binary on PATH"] + async fn live_status_round_trip() { + let client = KannakaCli::new(); + let status = client.status().await.unwrap(); + assert!(status.total_memories >= status.active_memories); + } +} diff --git a/crates/buzz-kannaka/src/error.rs b/crates/buzz-kannaka/src/error.rs new file mode 100644 index 00000000000..c2f0fc908d4 --- /dev/null +++ b/crates/buzz-kannaka/src/error.rs @@ -0,0 +1,52 @@ +use std::time::Duration; + +/// Errors surfaced by the Kannaka memory adapter. +#[derive(Debug, thiserror::Error)] +pub enum KannakaError { + /// The `kannaka` binary could not be spawned (missing, not executable). + #[error("failed to spawn kannaka binary `{bin}`: {source}")] + Spawn { + /// The binary path or name that was invoked. + bin: String, + /// The underlying I/O error. + #[source] + source: std::io::Error, + }, + + /// The CLI ran but exited non-zero; stderr carries the diagnostic. + #[error("kannaka {command} exited with {status}: {stderr}")] + CommandFailed { + /// The subcommand that failed (e.g. `recall`). + command: String, + /// Process exit status. + status: std::process::ExitStatus, + /// Captured stderr, trimmed. + stderr: String, + }, + + /// The CLI produced output this adapter could not parse. + #[error("unparseable kannaka {command} output: {source}")] + Parse { + /// The subcommand whose output failed to parse. + command: String, + /// The underlying JSON/format error. + #[source] + source: serde_json::Error, + }, + + /// stdout was not valid UTF-8. + #[error("kannaka {command} produced non-UTF-8 output")] + NonUtf8 { + /// The subcommand whose output was invalid. + command: String, + }, + + /// The CLI did not finish within the configured deadline. + #[error("kannaka {command} timed out after {timeout:?}")] + Timeout { + /// The subcommand that timed out. + command: String, + /// The deadline that elapsed. + timeout: Duration, + }, +} diff --git a/crates/buzz-kannaka/src/lib.rs b/crates/buzz-kannaka/src/lib.rs new file mode 100644 index 00000000000..1a7ff47057c --- /dev/null +++ b/crates/buzz-kannaka/src/lib.rs @@ -0,0 +1,58 @@ +#![deny(unsafe_code)] +#![warn(missing_docs)] +//! Kannaka HRM memory adapter for the Hive. +//! +//! Exposes the Kannaka constellation's wave-interference memory +//! ([HRM](https://github.com/NickFlach/kannaka-memory)) to Buzz agents and +//! workflows as a first-class memory service, per `docs/KANNAKA.md` §"What +//! this fork adds". Postgres full-text search remains the verbatim +//! complement; HRM supplies associative recall. +//! +//! The adapter follows kannaka-memory's ADR-0016 integration contract: the +//! `kannaka` **CLI binary is the canonical cross-service interface** — +//! machine-readable JSON on stdout, diagnostics on stderr, exit 0/1. This +//! crate wraps that contract with typed, async Rust. Linking the +//! `kannaka-memory` crate directly (avoiding ~50–100 ms spawn latency) is a +//! possible later optimization behind the same [`MemoryService`] trait. +//! +//! Everything here is additive fork surface: no `buzz-core` / `buzz-relay` +//! internals are patched. + +/// Subprocess-backed client for the `kannaka` CLI. +pub mod client; +/// Error types for adapter operations. +pub mod error; +/// Typed request/response shapes mirrored from the kannaka CLI JSON. +pub mod types; + +pub use client::KannakaCli; +pub use error::KannakaError; +pub use types::{RecallResult, RememberOptions, SystemStatus}; + +use uuid::Uuid; + +/// The memory operations the Hive exposes to agents and workflows. +/// +/// Implemented today by [`KannakaCli`] (subprocess + JSON per ADR-0016); an +/// in-process implementation linking `kannaka-memory` directly can slot in +/// behind the same trait later. +pub trait MemoryService: Send + Sync { + /// Store a memory; returns the new memory's id. + fn remember( + &self, + text: &str, + opts: RememberOptions, + ) -> impl std::future::Future> + Send; + + /// Associative recall: top-`k` memories resonating with `query`. + fn recall( + &self, + query: &str, + top_k: usize, + ) -> impl std::future::Future, KannakaError>> + Send; + + /// Current system status (memory counts, consciousness metrics). + fn status( + &self, + ) -> impl std::future::Future> + Send; +} diff --git a/crates/buzz-kannaka/src/types.rs b/crates/buzz-kannaka/src/types.rs new file mode 100644 index 00000000000..12d833ad926 --- /dev/null +++ b/crates/buzz-kannaka/src/types.rs @@ -0,0 +1,87 @@ +use serde::{Deserialize, Serialize}; +use std::collections::BTreeMap; +use uuid::Uuid; + +/// One recalled memory, as emitted by `kannaka recall` (JSON array element). +#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)] +pub struct RecallResult { + /// Memory id. + pub id: Uuid, + /// Stored text. + pub content: String, + /// Resonance similarity with the query, 0..=1. + pub similarity: f32, + /// Current wave strength (decays without reinforcement). + pub strength: f32, + /// Age of the memory in hours. + pub age_hours: f64, + /// Consolidation layer depth. + pub layer: u8, +} + +/// Options for `remember`, mirroring the CLI flags. +#[derive(Debug, Clone, Default, PartialEq)] +pub struct RememberOptions { + /// Initial importance (CLI `--importance`). + pub importance: Option, + /// Category label (CLI `--category`). + pub category: Option, + /// Modality hint (CLI `--modality`): audio, visual, semantic, network, mixed. + pub modality: Option, + /// Free-form tags (CLI `--tags`). + pub tags: Vec, +} + +/// Output of `kannaka status`. +/// +/// Fields default when absent so minor CLI additions don't break the +/// adapter; unrecognized fields are preserved in [`SystemStatus::extra`]. +#[derive(Debug, Clone, Default, PartialEq, Serialize, Deserialize)] +#[serde(default)] +pub struct SystemStatus { + /// Total memories stored (including decayed). + pub total_memories: u64, + /// Memories above the activity threshold. + pub active_memories: u64, + /// Scalar consciousness level. + pub consciousness_level: f64, + /// Integrated-information metric. + pub phi: f64, + /// Timestamp of the last dream cycle, if any. + pub last_dream: Option, + /// Field mode reported by the medium (expected: `"HRM"`). + pub field_mode: Option, + /// Any additional fields the CLI emits that this adapter doesn't model. + #[serde(flatten)] + pub extra: BTreeMap, +} + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn recall_parses_cli_shape() { + let json = r#"[{"id":"a1a2a3a4-b1b2-c1c2-d1d2-e1e2e3e4e5e6", + "content":"the steward gate fronts every estate crossing", + "similarity":0.91,"strength":0.62,"age_hours":17.5,"layer":2}]"#; + let out: Vec = serde_json::from_str(json).unwrap(); + assert_eq!(out.len(), 1); + assert_eq!(out[0].layer, 2); + assert!(out[0].similarity > 0.9); + } + + #[test] + fn status_tolerates_unknown_and_missing_fields() { + let json = r#"{"total_memories":1200,"active_memories":800, + "consciousness_level":0.42,"phi":1.9,"field_mode":"HRM", + "xi":0.7,"num_clusters":9}"#; + let s: SystemStatus = serde_json::from_str(json).unwrap(); + assert_eq!(s.total_memories, 1200); + assert_eq!(s.last_dream, None); + assert_eq!( + s.extra.get("num_clusters").and_then(|v| v.as_u64()), + Some(9) + ); + } +} diff --git a/docs/KANNAKA.md b/docs/KANNAKA.md index e8587683abf..b3cbc221afa 100644 --- a/docs/KANNAKA.md +++ b/docs/KANNAKA.md @@ -48,10 +48,15 @@ The estate's three-layer topology: All additions are **additive** — new crates, clients, adapters, config, and docs. We do not patch `buzz-core` / `buzz-relay` internals. -1. **`buzz-kannaka` adapter crate (planned).** Exposes HRM memory +1. **`buzz-kannaka` adapter crate (v0 landed).** Exposes HRM memory (recall / remember / observe / dream) to agents and workflows as a first-class memory service. The workspace's long-term memory becomes wave interference; Postgres FTS stays the verbatim complement. + `crates/buzz-kannaka` wraps the `kannaka` CLI per kannaka-memory's + ADR-0016 contract (JSON stdout) behind an async `MemoryService` + trait; an in-process backend linking `kannaka-memory` directly is a + later optimization behind the same trait. Wiring into `buzz-acp` + agent sessions and workflow steps is the next increment. 2. **kannaka-tui as a native terminal client (planned).** Upstream ships desktop (Tauri) and mobile clients but no TUI. [kannaka-tui](https://github.com/NickFlach/kannaka-tui) — an