kent/consciousness
1
0
Fork 1
Code Issues 1 Pull requests 5 Projects Releases Packages Wiki Activity Actions

CLI: convert node commands to typed async API

Browse source 
- node.rs: use memory::* typed helpers instead of memory_rpc()
- main.rs: make Run trait async, await all command dispatch
- defs.rs: bridge get_group_content async via block_in_place

Co-Authored-By: Proof of Concept <poc@bcachefs.org>
This commit is contained in:
Kent Overstreet 2026-04-13 13:20:04 -04:00
parent 933221f482
commit fa50f1c826
3 changed files with 84 additions and 102 deletions
Download patch file Download diff file Expand all files Collapse all files

133
src/cli/node.rs
View file

@ -3,53 +3,46 @@
// render, write, node-delete, node-rename, history, list-keys, // render, write, node-delete, node-rename, history, list-keys,
// list-edges, dump-json, lookup-bump, lookups. // list-edges, dump-json, lookup-bump, lookups.
use crate::agent::tools::memory;
use crate::store; use crate::store;
pub fn cmd_weight_set(key: &str, weight: f32) -> Result<(), String> { pub async fn cmd_weight_set(key: &str, weight: f32) -> Result<(), String> {
super::check_dry_run(); super::check_dry_run();
let result = crate::mcp_server::memory_rpc( let result = memory::weight_set(None, key, weight).await
"memory_weight_set", .map_err(|e| e.to_string())?;
serde_json::json!({"key": key, "weight": weight}),
).map_err(|e| e.to_string())?;
println!("{}", result); println!("{}", result);
Ok(()) Ok(())
} }
pub fn cmd_node_delete(key: &[String]) -> Result<(), String> { pub async fn cmd_node_delete(key: &[String]) -> Result<(), String> {
if key.is_empty() { if key.is_empty() {
return Err("node-delete requires a key".into()); return Err("node-delete requires a key".into());
} }
super::check_dry_run(); super::check_dry_run();
let key = key.join(" "); let key = key.join(" ");
let result = crate::mcp_server::memory_rpc( let result = memory::delete(None, &key).await
"memory_delete", .map_err(|e| e.to_string())?;
serde_json::json!({"key": key}),
).map_err(|e| e.to_string())?;
println!("{}", result); println!("{}", result);
Ok(()) Ok(())
} }
pub fn cmd_node_rename(old_key: &str, new_key: &str) -> Result<(), String> { pub async fn cmd_node_rename(old_key: &str, new_key: &str) -> Result<(), String> {
super::check_dry_run(); super::check_dry_run();
let result = crate::mcp_server::memory_rpc( let result = memory::rename(None, old_key, new_key).await
"memory_rename", .map_err(|e| e.to_string())?;
serde_json::json!({"old_key": old_key, "new_key": new_key}),
).map_err(|e| e.to_string())?;
println!("{}", result); println!("{}", result);
Ok(()) Ok(())
} }
pub fn cmd_render(key: &[String]) -> Result<(), String> { pub async fn cmd_render(key: &[String]) -> Result<(), String> {
if key.is_empty() { if key.is_empty() {
return Err("render requires a key".into()); return Err("render requires a key".into());
} }
let key = key.join(" "); let key = key.join(" ");
let bare = store::strip_md_suffix(&key); let bare = store::strip_md_suffix(&key);
let rendered = crate::mcp_server::memory_rpc( let rendered = memory::render(None, &bare, None).await
"memory_render", .map_err(|e| e.to_string())?;
serde_json::json!({"key": bare}),
).map_err(|e| e.to_string())?;
print!("{}", rendered); print!("{}", rendered);
// Mark as seen if we're inside a Claude session (not an agent subprocess — // Mark as seen if we're inside a Claude session (not an agent subprocess —
@ -73,20 +66,18 @@ pub fn cmd_render(key: &[String]) -> Result<(), String> {
Ok(()) Ok(())
} }
pub fn cmd_history(key: &[String], full: bool) -> Result<(), String> { pub async fn cmd_history(key: &[String], full: bool) -> Result<(), String> {
if key.is_empty() { if key.is_empty() {
return Err("history requires a key".into()); return Err("history requires a key".into());
} }
let key = key.join(" "); let key = key.join(" ");
let result = crate::mcp_server::memory_rpc( let result = memory::history(None, &key, Some(full)).await
"memory_history", .map_err(|e| e.to_string())?;
serde_json::json!({"key": key, "full": full}),
).map_err(|e| e.to_string())?;
print!("{}", result); print!("{}", result);
Ok(()) Ok(())
} }
pub fn cmd_write(key: &[String]) -> Result<(), String> { pub async fn cmd_write(key: &[String]) -> Result<(), String> {
if key.is_empty() { if key.is_empty() {
return Err("write requires a key (reads content from stdin)".into()); return Err("write requires a key (reads content from stdin)".into());
} }
@ -100,25 +91,21 @@ pub fn cmd_write(key: &[String]) -> Result<(), String> {
} }
super::check_dry_run(); super::check_dry_run();
let result = crate::mcp_server::memory_rpc( let result = memory::write(None, &key, &content).await
"memory_write", .map_err(|e| e.to_string())?;
serde_json::json!({"key": key, "content": content}),
).map_err(|e| e.to_string())?;
println!("{}", result); println!("{}", result);
Ok(()) Ok(())
} }
pub fn cmd_edit(key: &[String]) -> Result<(), String> { pub async fn cmd_edit(key: &[String]) -> Result<(), String> {
if key.is_empty() { if key.is_empty() {
return Err("edit requires a key".into()); return Err("edit requires a key".into());
} }
let key = key.join(" "); let key = key.join(" ");
// Get raw content via RPC // Get raw content
let content = crate::mcp_server::memory_rpc( let content = memory::render(None, &key, Some(true)).await
"memory_render", .unwrap_or_default();
serde_json::json!({"key": key, "raw": true}),
).unwrap_or_default();
let tmp = std::env::temp_dir().join(format!("poc-memory-edit-{}.md", key.replace('/', "_"))); let tmp = std::env::temp_dir().join(format!("poc-memory-edit-{}.md", key.replace('/', "_")));
std::fs::write(&tmp, &content) std::fs::write(&tmp, &content)
@ -149,42 +136,36 @@ pub fn cmd_edit(key: &[String]) -> Result<(), String> {
} }
super::check_dry_run(); super::check_dry_run();
let result = crate::mcp_server::memory_rpc( let result = memory::write(None, &key, &new_content).await
"memory_write", .map_err(|e| e.to_string())?;
serde_json::json!({"key": key, "content": new_content}),
).map_err(|e| e.to_string())?;
println!("{}", result); println!("{}", result);
Ok(()) Ok(())
} }
pub fn cmd_search(keys: &[String]) -> Result<(), String> { pub async fn cmd_search(keys: &[String]) -> Result<(), String> {
if keys.is_empty() { if keys.is_empty() {
return Err("search requires seed keys".into()); return Err("search requires seed keys".into());
} }
let result = crate::mcp_server::memory_rpc( let result = memory::search(None, keys.to_vec(), None, None, None, None).await
"memory_search", .map_err(|e| e.to_string())?;
serde_json::json!({"keys": keys}),
).map_err(|e| e.to_string())?;
print!("{}", result); print!("{}", result);
Ok(()) Ok(())
} }
pub fn cmd_query(expr: &[String]) -> Result<(), String> { pub async fn cmd_query(expr: &[String]) -> Result<(), String> {
if expr.is_empty() { if expr.is_empty() {
return Err("query requires an expression (try: poc-memory query --help)".into()); return Err("query requires an expression (try: poc-memory query --help)".into());
} }
let query_str = expr.join(" "); let query_str = expr.join(" ");
let result = crate::mcp_server::memory_rpc( let result = memory::query(None, &query_str, None).await
"memory_query", .map_err(|e| e.to_string())?;
serde_json::json!({"query": query_str}),
).map_err(|e| e.to_string())?;
print!("{}", result); print!("{}", result);
Ok(()) Ok(())
} }
/// Get group content via RPC (handles daemon or local fallback) /// Get group content (handles daemon or local fallback)
pub fn get_group_content(group: &crate::config::ContextGroup, cfg: &crate::config::Config) -> Vec<(String, String)> { pub async fn get_group_content(group: &crate::config::ContextGroup, cfg: &crate::config::Config) -> Vec<(String, String)> {
match group.source { match group.source {
crate::config::ContextSource::Journal => { crate::config::ContextSource::Journal => {
// Query for recent journal entries // Query for recent journal entries
@ -192,26 +173,21 @@ pub fn get_group_content(group: &crate::config::ContextGroup, cfg: &crate::confi
let query = format!("all | type:episodic | age:<{} | sort:timestamp | limit:{}", let query = format!("all | type:episodic | age:<{} | sort:timestamp | limit:{}",
window, cfg.journal_max); window, cfg.journal_max);
let keys_str = match crate::mcp_server::memory_rpc( let keys_str = match memory::query(None, &query, None).await {
"memory_query",
serde_json::json!({"query": query}),
) {
Ok(s) => s, Ok(s) => s,
Err(_) => return vec![], Err(_) => return vec![],
}; };
// Parse keys (one per line) and render each // Parse keys (one per line) and render each
keys_str.lines() let mut results = Vec::new();
.filter(|k| !k.is_empty() && *k != "no results") for key in keys_str.lines().filter(|k| !k.is_empty() && *k != "no results") {
.filter_map(|key| { if let Ok(content) = memory::render(None, key, Some(true)).await {
let content = crate::mcp_server::memory_rpc( if !content.trim().is_empty() {
"memory_render", results.push((key.to_string(), content));
serde_json::json!({"key": key, "raw": true}), }
).ok()?; }
if content.trim().is_empty() { return None; } }
Some((key.to_string(), content)) results
})
.collect()
} }
crate::config::ContextSource::File => { crate::config::ContextSource::File => {
group.keys.iter().filter_map(|key| { group.keys.iter().filter_map(|key| {
@ -221,19 +197,20 @@ pub fn get_group_content(group: &crate::config::ContextGroup, cfg: &crate::confi
}).collect() }).collect()
} }
crate::config::ContextSource::Store => { crate::config::ContextSource::Store => {
group.keys.iter().filter_map(|key| { let mut results = Vec::new();
let content = crate::mcp_server::memory_rpc( for key in &group.keys {
"memory_render", if let Ok(content) = memory::render(None, key, Some(true)).await {
serde_json::json!({"key": key, "raw": true}), if !content.trim().is_empty() {
).ok()?; results.push((key.clone(), content.trim().to_string()));
if content.trim().is_empty() { return None; } }
Some((key.clone(), content.trim().to_string())) }
}).collect() }
results
} }
} }
} }
pub fn cmd_load_context(stats: bool) -> Result<(), String> { pub async fn cmd_load_context(stats: bool) -> Result<(), String> {
let cfg = crate::config::get(); let cfg = crate::config::get();
if stats { if stats {
@ -243,7 +220,7 @@ pub fn cmd_load_context(stats: bool) -> Result<(), String> {
println!("{}", "-".repeat(42)); println!("{}", "-".repeat(42));
for group in &cfg.context_groups { for group in &cfg.context_groups {
let entries = get_group_content(group, &cfg); let entries = get_group_content(group, &cfg).await;
let words: usize = entries.iter() let words: usize = entries.iter()
.map(|(_, c)| c.split_whitespace().count()) .map(|(_, c)| c.split_whitespace().count())
.sum(); .sum();
@ -261,7 +238,7 @@ pub fn cmd_load_context(stats: bool) -> Result<(), String> {
println!("=== MEMORY SYSTEM ({}) ===", cfg.assistant_name); println!("=== MEMORY SYSTEM ({}) ===", cfg.assistant_name);
for group in &cfg.context_groups { for group in &cfg.context_groups {
let entries = get_group_content(group, &cfg); let entries = get_group_content(group, &cfg).await;
if !entries.is_empty() && group.source == crate::config::ContextSource::Journal { if !entries.is_empty() && group.source == crate::config::ContextSource::Journal {
println!("--- recent journal entries ({}/{}) ---", println!("--- recent journal entries ({}/{}) ---",
entries.len(), cfg.journal_max); entries.len(), cfg.journal_max);

46
src/main.rs
View file

@ -384,43 +384,43 @@ fn print_help() {
// ── Dispatch ───────────────────────────────────────────────────────── // ── Dispatch ─────────────────────────────────────────────────────────
trait Run { trait Run {
fn run(self) -> Result<(), String>; async fn run(self) -> Result<(), String>;
} }
impl Run for Command { impl Run for Command {
fn run(self) -> Result<(), String> { async fn run(self) -> Result<(), String> {
match self { match self {
Self::Search { keys } => cli::node::cmd_search(&keys), Self::Search { keys } => cli::node::cmd_search(&keys).await,
Self::Render { key } => cli::node::cmd_render(&key), Self::Render { key } => cli::node::cmd_render(&key).await,
Self::Write { key } => cli::node::cmd_write(&key), Self::Write { key } => cli::node::cmd_write(&key).await,
Self::Edit { key } => cli::node::cmd_edit(&key), Self::Edit { key } => cli::node::cmd_edit(&key).await,
Self::History { full, key } => cli::node::cmd_history(&key, full), Self::History { full, key } => cli::node::cmd_history(&key, full).await,
Self::Tail { n, full, provenance, all_versions } Self::Tail { n, full, provenance, all_versions }
=> cli::journal::cmd_tail(n, full, provenance.as_deref(), !all_versions), => cli::journal::cmd_tail(n, full, provenance.as_deref(), !all_versions),
Self::Status => cli::admin::cmd_status(), Self::Status => cli::admin::cmd_status(),
Self::Query { expr } => cli::node::cmd_query(&expr), Self::Query { expr } => cli::node::cmd_query(&expr).await,
Self::WeightSet { key, weight } => cli::node::cmd_weight_set(&key, weight), Self::WeightSet { key, weight } => cli::node::cmd_weight_set(&key, weight).await,
Self::Node(sub) => sub.run(), Self::Node(sub) => sub.run().await,
Self::Journal(sub) => sub.run(), Self::Journal(sub) => sub.run().await,
Self::GraphCmd(sub) => sub.run(), Self::GraphCmd(sub) => sub.run().await,
Self::Agent(sub) => sub.run(), Self::Agent(sub) => sub.run().await,
Self::Admin(sub) => sub.run(), Self::Admin(sub) => sub.run().await,
// mcp-schema moved to consciousness-mcp binary // mcp-schema moved to consciousness-mcp binary
} }
} }
} }
impl Run for NodeCmd { impl Run for NodeCmd {
fn run(self) -> Result<(), String> { async fn run(self) -> Result<(), String> {
match self { match self {
Self::Delete { key } => cli::node::cmd_node_delete(&key), Self::Delete { key } => cli::node::cmd_node_delete(&key).await,
Self::Rename { old_key, new_key } => cli::node::cmd_node_rename(&old_key, &new_key), Self::Rename { old_key, new_key } => cli::node::cmd_node_rename(&old_key, &new_key).await,
} }
} }
} }
impl Run for JournalCmd { impl Run for JournalCmd {
fn run(self) -> Result<(), String> { async fn run(self) -> Result<(), String> {
match self { match self {
Self::Write { name, text } => cli::journal::cmd_journal_write(&name, &text), Self::Write { name, text } => cli::journal::cmd_journal_write(&name, &text),
Self::Tail { n, full, level } => cli::journal::cmd_journal_tail(n, full, level), Self::Tail { n, full, level } => cli::journal::cmd_journal_tail(n, full, level),
@ -429,7 +429,7 @@ impl Run for JournalCmd {
} }
impl Run for GraphCmd { impl Run for GraphCmd {
fn run(self) -> Result<(), String> { async fn run(self) -> Result<(), String> {
match self { match self {
Self::Link { key } => cli::graph::cmd_link(&key), Self::Link { key } => cli::graph::cmd_link(&key),
Self::LinkAdd { source, target, reason } Self::LinkAdd { source, target, reason }
@ -446,7 +446,7 @@ impl Run for GraphCmd {
} }
impl Run for AgentCmd { impl Run for AgentCmd {
fn run(self) -> Result<(), String> { async fn run(self) -> Result<(), String> {
match self { match self {
Self::Run { agent, count, target, query, dry_run, local, state_dir } Self::Run { agent, count, target, query, dry_run, local, state_dir }
=> cli::agent::cmd_run_agent(&agent, count, &target, query.as_deref(), dry_run, local, state_dir.as_deref()), => cli::agent::cmd_run_agent(&agent, count, &target, query.as_deref(), dry_run, local, state_dir.as_deref()),
@ -455,7 +455,7 @@ impl Run for AgentCmd {
} }
impl Run for AdminCmd { impl Run for AdminCmd {
fn run(self) -> Result<(), String> { async fn run(self) -> Result<(), String> {
match self { match self {
Self::Init => cli::admin::cmd_init(), Self::Init => cli::admin::cmd_init(),
Self::Health => cli::admin::cmd_health(), Self::Health => cli::admin::cmd_health(),
@ -465,7 +465,7 @@ impl Run for AdminCmd {
Self::DailyCheck => cli::admin::cmd_daily_check(), Self::DailyCheck => cli::admin::cmd_daily_check(),
Self::Import { files } => cli::admin::cmd_import(&files), Self::Import { files } => cli::admin::cmd_import(&files),
Self::Export { files, all } => cli::admin::cmd_export(&files, all), Self::Export { files, all } => cli::admin::cmd_export(&files, all),
Self::LoadContext { stats } => cli::node::cmd_load_context(stats), Self::LoadContext { stats } => cli::node::cmd_load_context(stats).await,
Self::MigrateTranscriptProgress => { Self::MigrateTranscriptProgress => {
let mut store = store::Store::load()?; let mut store = store::Store::load()?;
let count = store.migrate_transcript_progress()?; let count = store.migrate_transcript_progress()?;
@ -496,7 +496,7 @@ async fn main() {
let cli = Cli::parse(); let cli = Cli::parse();
if let Err(e) = cli.command.run() { if let Err(e) = cli.command.run().await {
eprintln!("Error: {}", e); eprintln!("Error: {}", e);
process::exit(1); process::exit(1);
} }

7
src/subconscious/defs.rs
View file

@ -303,7 +303,12 @@ fn resolve(
let mut keys = Vec::new(); let mut keys = Vec::new();
for group in &cfg.context_groups { for group in &cfg.context_groups {
if !group.agent { continue; } if !group.agent { continue; }
let entries = crate::cli::node::get_group_content(group, &cfg); // Bridge sync→async using block_in_place (same as resolve_tool)
let entries = tokio::task::block_in_place(|| {
tokio::runtime::Handle::current().block_on(
crate::cli::node::get_group_content(group, &cfg)
)
});
for (key, content) in entries { for (key, content) in entries {
use std::fmt::Write; use std::fmt::Write;
writeln!(text, "--- {} ({}) ---", key, group.label).ok(); writeln!(text, "--- {} ({}) ---", key, group.label).ok();