diff --git a/app/src-tauri/Cargo.lock b/app/src-tauri/Cargo.lock index f93b5f31d..0ea31645a 100644 --- a/app/src-tauri/Cargo.lock +++ b/app/src-tauri/Cargo.lock @@ -4,7 +4,7 @@ version = 4 [[package]] name = "OpenHuman" -version = "0.51.17" +version = "0.51.18" dependencies = [ "env_logger", "log", diff --git a/src/core/all.rs b/src/core/all.rs index c5fdd2613..fcfeca2ed 100644 --- a/src/core/all.rs +++ b/src/core/all.rs @@ -74,6 +74,8 @@ fn build_registered_controllers() -> Vec { controllers.extend(crate::openhuman::webhooks::all_webhooks_registered_controllers()); controllers.extend(crate::openhuman::update::all_update_registered_controllers()); controllers + .extend(crate::openhuman::tree_summarizer::all_tree_summarizer_registered_controllers()); + controllers } fn build_declared_controller_schemas() -> Vec { @@ -111,6 +113,7 @@ fn build_declared_controller_schemas() -> Vec { schemas.extend(crate::openhuman::subconscious::all_subconscious_controller_schemas()); schemas.extend(crate::openhuman::webhooks::all_webhooks_controller_schemas()); schemas.extend(crate::openhuman::update::all_update_controller_schemas()); + schemas.extend(crate::openhuman::tree_summarizer::all_tree_summarizer_controller_schemas()); schemas } @@ -158,6 +161,9 @@ pub fn namespace_description(namespace: &str) -> Option<&'static str> { "update" => { Some("Self-update: check GitHub Releases for newer core binary and stage updates.") } + "tree_summarizer" => { + Some("Hierarchical time-based summarization tree for background knowledge compression.") + } _ => None, } } diff --git a/src/core/cli.rs b/src/core/cli.rs index 538956234..15375bdc2 100644 --- a/src/core/cli.rs +++ b/src/core/cli.rs @@ -61,6 +61,9 @@ pub fn run_from_cli_args(args: &[String]) -> Result<()> { } "voice" | "dictate" => run_voice_server_command(&args[1..]), "text-input" => crate::core::text_input_cli::run_text_input_command(&args[1..]), + "tree-summarizer" => { + crate::core::tree_summarizer_cli::run_tree_summarizer_command(&args[1..]) + } "memory" => crate::core::memory_cli::run_memory_command(&args[1..]), namespace => run_namespace_command(namespace, &args[1..], &grouped), } @@ -479,6 +482,7 @@ fn print_general_help(grouped: &BTreeMap>) { println!(" openhuman call --method [--params '']"); println!(" openhuman skills [options] (skill development runtime)"); println!(" openhuman voice [--hotkey ] [--mode ] (voice dictation server)"); + println!(" openhuman tree-summarizer [options] (summary tree CLI)"); println!(" openhuman [--param value ...]\n"); println!("Available namespaces:"); for namespace in grouped.keys() { diff --git a/src/core/mod.rs b/src/core/mod.rs index 65aa62bc7..641f46a55 100644 --- a/src/core/mod.rs +++ b/src/core/mod.rs @@ -14,6 +14,7 @@ pub mod screen_intelligence_cli; pub mod skills_cli; pub mod socketio; pub mod text_input_cli; +pub mod tree_summarizer_cli; pub mod types; /// Canonical function contract for domain controllers. diff --git a/src/core/tree_summarizer_cli.rs b/src/core/tree_summarizer_cli.rs new file mode 100644 index 000000000..9ddc4f30c --- /dev/null +++ b/src/core/tree_summarizer_cli.rs @@ -0,0 +1,386 @@ +//! `openhuman tree-summarizer` — CLI for the hierarchical summary tree. +//! +//! Ingest content, run summarization jobs, query the tree, and inspect +//! status from the terminal without starting the full app. +//! +//! Usage: +//! openhuman tree-summarizer ingest [--content | --file ] [-v] +//! openhuman tree-summarizer run [-v] +//! openhuman tree-summarizer query [] [-v] +//! openhuman tree-summarizer status [-v] +//! openhuman tree-summarizer rebuild [-v] + +use anyhow::Result; + +/// Entry point for `openhuman tree-summarizer `. +pub fn run_tree_summarizer_command(args: &[String]) -> Result<()> { + if args.is_empty() || is_help(&args[0]) { + print_help(); + return Ok(()); + } + + match args[0].as_str() { + "ingest" => run_ingest(&args[1..]), + "run" => run_summarize(&args[1..]), + "query" => run_query(&args[1..]), + "status" => run_status(&args[1..]), + "rebuild" => run_rebuild(&args[1..]), + other => Err(anyhow::anyhow!( + "unknown tree-summarizer subcommand '{other}'. Run `openhuman tree-summarizer --help`." + )), + } +} + +// --------------------------------------------------------------------------- +// Option parsing +// --------------------------------------------------------------------------- + +struct CliOpts { + verbose: bool, + content: Option, + file: Option, + node_id: Option, +} + +fn parse_opts(args: &[String]) -> Result<(CliOpts, Vec)> { + let mut verbose = false; + let mut content: Option = None; + let mut file: Option = None; + let mut node_id: Option = None; + let mut rest = Vec::new(); + let mut i = 0; + + while i < args.len() { + match args[i].as_str() { + "--content" | "-c" => { + let val = args + .get(i + 1) + .ok_or_else(|| anyhow::anyhow!("missing value for --content"))?; + content = Some(val.clone()); + i += 2; + } + "--file" | "-f" => { + let val = args + .get(i + 1) + .ok_or_else(|| anyhow::anyhow!("missing value for --file"))?; + file = Some(val.clone()); + i += 2; + } + "--node-id" | "--node" => { + let val = args + .get(i + 1) + .ok_or_else(|| anyhow::anyhow!("missing value for --node-id"))?; + node_id = Some(val.clone()); + i += 2; + } + "-v" | "--verbose" => { + verbose = true; + i += 1; + } + "-h" | "--help" => { + rest.push(args[i].clone()); + i += 1; + } + _ => { + rest.push(args[i].clone()); + i += 1; + } + } + } + + Ok(( + CliOpts { + verbose, + content, + file, + node_id, + }, + rest, + )) +} + +// --------------------------------------------------------------------------- +// Subcommands +// --------------------------------------------------------------------------- + +/// `openhuman tree-summarizer ingest --content ` or `--file ` +fn run_ingest(args: &[String]) -> Result<()> { + let (opts, rest) = parse_opts(args)?; + + if rest.iter().any(|a| is_help(a)) || rest.is_empty() { + println!("Usage: openhuman tree-summarizer ingest [--content ] [--file ] [-v]"); + println!(); + println!("Append content to the summarization buffer for a namespace."); + println!(); + println!(" Target namespace for the summary tree"); + println!(" --content, -c Raw text content to ingest"); + println!(" --file, -f Read content from a file (use - for stdin)"); + println!(" -v, --verbose Enable debug logging"); + println!(); + println!("Either --content or --file is required. If both are given, --file wins."); + return Ok(()); + } + + let namespace = &rest[0]; + + let content = if let Some(ref path) = opts.file { + if path == "-" { + use std::io::Read; + let mut buf = String::new(); + std::io::stdin() + .read_to_string(&mut buf) + .map_err(|e| anyhow::anyhow!("failed to read stdin: {e}"))?; + buf + } else { + std::fs::read_to_string(path) + .map_err(|e| anyhow::anyhow!("failed to read '{}': {e}", path))? + } + } else if let Some(ref text) = opts.content { + text.clone() + } else { + return Err(anyhow::anyhow!( + "either --content or --file is required. Run `openhuman tree-summarizer ingest --help`." + )); + }; + + if content.trim().is_empty() { + return Err(anyhow::anyhow!("content is empty")); + } + + init_logging(opts.verbose); + + let rt = build_runtime()?; + rt.block_on(async { + let config = load_config().await?; + let outcome = crate::openhuman::tree_summarizer::rpc::tree_summarizer_ingest( + &config, namespace, &content, None, None, + ) + .await + .map_err(anyhow::Error::msg)?; + + println!( + "{}", + serde_json::to_string_pretty(&outcome.value) + .unwrap_or_else(|_| format!("{:?}", outcome.value)) + ); + Ok(()) + }) +} + +/// `openhuman tree-summarizer run ` +fn run_summarize(args: &[String]) -> Result<()> { + let (opts, rest) = parse_opts(args)?; + + if rest.iter().any(|a| is_help(a)) || rest.is_empty() { + println!("Usage: openhuman tree-summarizer run [-v]"); + println!(); + println!("Trigger the summarization job for a namespace."); + println!("Drains the buffer, creates the hour leaf, and propagates upward."); + println!(); + println!(" Target namespace"); + println!(" -v, --verbose Enable debug logging"); + return Ok(()); + } + + let namespace = &rest[0]; + init_logging(opts.verbose); + + let rt = build_runtime()?; + rt.block_on(async { + let config = load_config().await?; + let outcome = + crate::openhuman::tree_summarizer::rpc::tree_summarizer_run(&config, namespace) + .await + .map_err(anyhow::Error::msg)?; + + println!( + "{}", + serde_json::to_string_pretty(&outcome.value) + .unwrap_or_else(|_| format!("{:?}", outcome.value)) + ); + Ok(()) + }) +} + +/// `openhuman tree-summarizer query []` +fn run_query(args: &[String]) -> Result<()> { + let (opts, rest) = parse_opts(args)?; + + if rest.iter().any(|a| is_help(a)) || rest.is_empty() { + println!( + "Usage: openhuman tree-summarizer query [] [--node-id ] [-v]" + ); + println!(); + println!("Read a summary tree node and its direct children."); + println!(); + println!(" Target namespace"); + println!(" Node ID to query (default: root)"); + println!(" --node-id, --node Alternative way to specify the node ID"); + println!(" -v, --verbose Enable debug logging"); + println!(); + println!("Node ID examples:"); + println!(" root All-time summary"); + println!(" 2024 Year summary"); + println!(" 2024/03 Month summary"); + println!(" 2024/03/15 Day summary"); + println!(" 2024/03/15/14 Hour leaf (2pm)"); + return Ok(()); + } + + let namespace = &rest[0]; + let node_id = opts + .node_id + .as_deref() + .or_else(|| rest.get(1).map(|s| s.as_str())); + + init_logging(opts.verbose); + + let rt = build_runtime()?; + rt.block_on(async { + let config = load_config().await?; + let outcome = crate::openhuman::tree_summarizer::rpc::tree_summarizer_query( + &config, namespace, node_id, + ) + .await + .map_err(anyhow::Error::msg)?; + + println!( + "{}", + serde_json::to_string_pretty(&outcome.value) + .unwrap_or_else(|_| format!("{:?}", outcome.value)) + ); + Ok(()) + }) +} + +/// `openhuman tree-summarizer status ` +fn run_status(args: &[String]) -> Result<()> { + let (opts, rest) = parse_opts(args)?; + + if rest.iter().any(|a| is_help(a)) || rest.is_empty() { + println!("Usage: openhuman tree-summarizer status [-v]"); + println!(); + println!("Show tree metadata: node count, depth, date range."); + println!(); + println!(" Target namespace"); + println!(" -v, --verbose Enable debug logging"); + return Ok(()); + } + + let namespace = &rest[0]; + init_logging(opts.verbose); + + let rt = build_runtime()?; + rt.block_on(async { + let config = load_config().await?; + let outcome = + crate::openhuman::tree_summarizer::rpc::tree_summarizer_status(&config, namespace) + .await + .map_err(anyhow::Error::msg)?; + + println!( + "{}", + serde_json::to_string_pretty(&outcome.value) + .unwrap_or_else(|_| format!("{:?}", outcome.value)) + ); + Ok(()) + }) +} + +/// `openhuman tree-summarizer rebuild ` +fn run_rebuild(args: &[String]) -> Result<()> { + let (opts, rest) = parse_opts(args)?; + + if rest.iter().any(|a| is_help(a)) || rest.is_empty() { + println!("Usage: openhuman tree-summarizer rebuild [-v]"); + println!(); + println!("Rebuild the entire summary tree from hour leaves upward."); + println!("This re-summarizes all intermediate levels (day, month, year, root)."); + println!(); + println!(" Target namespace"); + println!(" -v, --verbose Enable debug logging"); + return Ok(()); + } + + let namespace = &rest[0]; + init_logging(opts.verbose); + + eprintln!(" Rebuilding tree for namespace '{namespace}'... this may take a while."); + + let rt = build_runtime()?; + rt.block_on(async { + let config = load_config().await?; + let outcome = + crate::openhuman::tree_summarizer::rpc::tree_summarizer_rebuild(&config, namespace) + .await + .map_err(anyhow::Error::msg)?; + + println!( + "{}", + serde_json::to_string_pretty(&outcome.value) + .unwrap_or_else(|_| format!("{:?}", outcome.value)) + ); + Ok(()) + }) +} + +// --------------------------------------------------------------------------- +// Helpers +// --------------------------------------------------------------------------- + +fn build_runtime() -> Result { + tokio::runtime::Builder::new_multi_thread() + .enable_all() + .build() + .map_err(|e| anyhow::anyhow!("failed to build tokio runtime: {e}")) +} + +async fn load_config() -> Result { + let mut config = crate::openhuman::config::Config::load_or_init() + .await + .unwrap_or_default(); + config.apply_env_overrides(); + Ok(config) +} + +fn init_logging(verbose: bool) { + if !verbose && std::env::var_os("RUST_LOG").is_none() { + unsafe { std::env::set_var("RUST_LOG", "warn") }; + } + crate::core::logging::init_for_cli_run(verbose, crate::core::logging::CliLogDefault::Global); +} + +fn is_help(value: &str) -> bool { + matches!(value, "-h" | "--help" | "help") +} + +fn print_help() { + println!("openhuman tree-summarizer — hierarchical summary tree\n"); + println!("Usage:"); + println!( + " openhuman tree-summarizer ingest [--content ] [--file ] [-v]" + ); + println!(" openhuman tree-summarizer run [-v]"); + println!(" openhuman tree-summarizer query [] [-v]"); + println!(" openhuman tree-summarizer status [-v]"); + println!(" openhuman tree-summarizer rebuild [-v]"); + println!(); + println!("Subcommands:"); + println!(" ingest Buffer raw content for the next summarization run"); + println!(" run Drain buffer → create hour leaf → propagate summaries upward"); + println!(" query Read a node and its children (default: root)"); + println!(" status Show tree metadata (node count, depth, date range)"); + println!(" rebuild Rebuild entire tree from hour leaves (re-summarizes all levels)"); + println!(); + println!("Common options:"); + println!(" -v, --verbose Enable debug logging"); + println!(); + println!("Examples:"); + println!(" openhuman tree-summarizer ingest my-ns --content 'Some raw data to summarize'"); + println!(" openhuman tree-summarizer ingest my-ns --file notes.txt"); + println!(" cat journal.md | openhuman tree-summarizer ingest my-ns --file -"); + println!(" openhuman tree-summarizer run my-ns"); + println!(" openhuman tree-summarizer query my-ns root"); + println!(" openhuman tree-summarizer query my-ns 2024/03/15"); + println!(" openhuman tree-summarizer status my-ns"); +} diff --git a/src/openhuman/channels/runtime/startup.rs b/src/openhuman/channels/runtime/startup.rs index 3894393c6..825eaefe1 100644 --- a/src/openhuman/channels/runtime/startup.rs +++ b/src/openhuman/channels/runtime/startup.rs @@ -426,6 +426,10 @@ pub async fn start_channels(config: Config) -> Result<()> { let _cron_delivery_handle = bus.subscribe(Arc::new( crate::openhuman::cron::bus::CronDeliverySubscriber::new(Arc::clone(&channels_by_name)), )); + // Register the tree summarizer event subscriber for observability logging. + let _tree_summarizer_handle = bus.subscribe(Arc::new( + crate::openhuman::tree_summarizer::bus::TreeSummarizerEventSubscriber::new(), + )); let max_in_flight_messages = compute_max_in_flight_messages(channels.len()); diff --git a/src/openhuman/event_bus/events.rs b/src/openhuman/event_bus/events.rs index f6de061ec..f6c1d13f6 100644 --- a/src/openhuman/event_bus/events.rs +++ b/src/openhuman/event_bus/events.rs @@ -142,6 +142,23 @@ pub enum DomainEvent { error: Option, }, + // ── Tree Summarizer ────────────────────────────────────────────────── + /// An hour leaf was created from buffered data. + TreeSummarizerHourCompleted { + namespace: String, + node_id: String, + token_count: u32, + }, + /// A tree node summary was updated during propagation. + TreeSummarizerPropagated { + namespace: String, + node_id: String, + level: String, + token_count: u32, + }, + /// A full tree rebuild completed. + TreeSummarizerRebuildCompleted { namespace: String, total_nodes: u64 }, + // ── System lifecycle ──────────────────────────────────────────────── /// A system component started up. SystemStartup { component: String }, @@ -184,6 +201,10 @@ impl DomainEvent { | Self::WebhookUnregistered { .. } | Self::WebhookProcessed { .. } => "webhook", + Self::TreeSummarizerHourCompleted { .. } + | Self::TreeSummarizerPropagated { .. } + | Self::TreeSummarizerRebuildCompleted { .. } => "tree_summarizer", + Self::SystemStartup { .. } | Self::SystemShutdown { .. } | Self::HealthChanged { .. } => "system", @@ -410,6 +431,31 @@ mod tests { }, "webhook", ), + // Tree Summarizer + ( + DomainEvent::TreeSummarizerHourCompleted { + namespace: "n".into(), + node_id: "2024/03/15/14".into(), + token_count: 500, + }, + "tree_summarizer", + ), + ( + DomainEvent::TreeSummarizerPropagated { + namespace: "n".into(), + node_id: "2024/03/15".into(), + level: "day".into(), + token_count: 1000, + }, + "tree_summarizer", + ), + ( + DomainEvent::TreeSummarizerRebuildCompleted { + namespace: "n".into(), + total_nodes: 10, + }, + "tree_summarizer", + ), // System ( DomainEvent::SystemStartup { diff --git a/src/openhuman/mod.rs b/src/openhuman/mod.rs index 502f42d07..1c06c3c2c 100644 --- a/src/openhuman/mod.rs +++ b/src/openhuman/mod.rs @@ -49,6 +49,7 @@ pub mod team; pub mod text_input; pub mod tool_timeout; pub mod tools; +pub mod tree_summarizer; pub mod update; pub mod util; pub mod voice; diff --git a/src/openhuman/tree_summarizer/bus.rs b/src/openhuman/tree_summarizer/bus.rs new file mode 100644 index 000000000..3d32f4ee9 --- /dev/null +++ b/src/openhuman/tree_summarizer/bus.rs @@ -0,0 +1,125 @@ +//! Event bus integration for tree_summarizer. +//! +//! Subscribes to `TreeSummarizer*` events and logs them for observability. +//! Future subscribers can react to these events for cross-module workflows. + +use crate::openhuman::event_bus::{DomainEvent, EventHandler}; +use async_trait::async_trait; + +/// Subscribes to tree summarizer events and logs activity. +pub struct TreeSummarizerEventSubscriber; + +impl TreeSummarizerEventSubscriber { + pub fn new() -> Self { + Self + } +} + +#[async_trait] +impl EventHandler for TreeSummarizerEventSubscriber { + fn name(&self) -> &str { + "tree_summarizer::events" + } + + fn domains(&self) -> Option<&[&str]> { + Some(&["tree_summarizer"]) + } + + async fn handle(&self, event: &DomainEvent) { + match event { + DomainEvent::TreeSummarizerHourCompleted { + namespace, + node_id, + token_count, + } => { + tracing::info!( + namespace = %namespace, + node_id = %node_id, + token_count = %token_count, + "[tree_summarizer] hour leaf completed" + ); + } + DomainEvent::TreeSummarizerPropagated { + namespace, + node_id, + level, + token_count, + } => { + tracing::info!( + namespace = %namespace, + node_id = %node_id, + level = %level, + token_count = %token_count, + "[tree_summarizer] node propagated" + ); + } + DomainEvent::TreeSummarizerRebuildCompleted { + namespace, + total_nodes, + } => { + tracing::info!( + namespace = %namespace, + total_nodes = %total_nodes, + "[tree_summarizer] tree rebuild completed" + ); + } + _ => {} + } + } +} + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn subscriber_name_and_domain() { + let sub = TreeSummarizerEventSubscriber::new(); + assert_eq!(sub.name(), "tree_summarizer::events"); + assert_eq!(sub.domains(), Some(&["tree_summarizer"][..])); + } + + #[tokio::test] + async fn handles_hour_completed_without_panic() { + let sub = TreeSummarizerEventSubscriber::new(); + sub.handle(&DomainEvent::TreeSummarizerHourCompleted { + namespace: "test".into(), + node_id: "2024/03/15/14".into(), + token_count: 500, + }) + .await; + } + + #[tokio::test] + async fn handles_propagated_without_panic() { + let sub = TreeSummarizerEventSubscriber::new(); + sub.handle(&DomainEvent::TreeSummarizerPropagated { + namespace: "test".into(), + node_id: "2024/03/15".into(), + level: "day".into(), + token_count: 1500, + }) + .await; + } + + #[tokio::test] + async fn handles_rebuild_without_panic() { + let sub = TreeSummarizerEventSubscriber::new(); + sub.handle(&DomainEvent::TreeSummarizerRebuildCompleted { + namespace: "test".into(), + total_nodes: 42, + }) + .await; + } + + #[tokio::test] + async fn ignores_unrelated_events() { + let sub = TreeSummarizerEventSubscriber::new(); + sub.handle(&DomainEvent::CronJobTriggered { + job_id: "j1".into(), + job_type: "shell".into(), + }) + .await; + // No panic = pass + } +} diff --git a/src/openhuman/tree_summarizer/engine.rs b/src/openhuman/tree_summarizer/engine.rs new file mode 100644 index 000000000..a949e1d07 --- /dev/null +++ b/src/openhuman/tree_summarizer/engine.rs @@ -0,0 +1,588 @@ +//! Core summarization engine: ingest raw data, summarize into hour leaves, +//! and propagate summaries upward through the tree. + +use anyhow::{Context, Result}; +use chrono::{DateTime, Timelike, Utc}; +use std::collections::BTreeMap; + +use crate::openhuman::config::Config; +use crate::openhuman::event_bus::{publish_global, DomainEvent}; +use crate::openhuman::providers::traits::Provider; +use crate::openhuman::tree_summarizer::store; +use crate::openhuman::tree_summarizer::types::{ + derive_node_ids, derive_parent_id, estimate_tokens, level_from_node_id, NodeLevel, TreeNode, + TreeStatus, +}; + +/// The model hint passed to the provider for summarization tasks. +const SUMMARIZATION_MODEL: &str = "hint:fast"; +const SUMMARIZATION_TEMP: f64 = 0.3; + +/// Maximum characters for a summary response (hard limit enforced after LLM call). +/// Set to 4x the Root token budget as a generous upper bound. +const MAX_SUMMARY_CHARS: usize = 20_000 * 4; + +// ── Public API ───────────────────────────────────────────────────────── + +/// Run the summarization job for a given namespace. +/// +/// 1. Drains the ingestion buffer. +/// 2. Groups buffered entries by their original hour (from filename timestamps). +/// 3. Summarizes each hour group into its own hour leaf. +/// 4. Propagates summaries upward through day → month → year → root. +/// +/// Returns the last hour leaf node created, or `None` if the buffer was empty. +pub async fn run_summarization( + config: &Config, + provider: &dyn Provider, + namespace: &str, + _ts: DateTime, +) -> Result> { + // Read buffer entries non-destructively; we only delete after durable writes. + let buffered = store::buffer_read(config, namespace)?; + if buffered.is_empty() { + tracing::debug!("[tree_summarizer] no buffered data for namespace '{namespace}', skipping"); + return Ok(None); + } + + let buffer_filenames: Vec = buffered.iter().map(|(name, _)| name.clone()).collect(); + + tracing::debug!( + "[tree_summarizer] starting summarization for namespace '{}' with {} buffer entries", + namespace, + buffered.len() + ); + + // Group buffered entries by hour using their buffer filename timestamps. + let hour_groups = group_by_hour(&buffered); + + tracing::debug!( + "[tree_summarizer] grouped into {} distinct hours", + hour_groups.len() + ); + + // Track all ancestor IDs to propagate after all hour leaves are written. + let mut all_propagation_ids: Vec<(String, NodeLevel)> = Vec::new(); + let mut last_hour_node: Option = None; + + for (hour_id, entries) in &hour_groups { + let combined = entries.join("\n\n---\n\n"); + + // Check for an existing hour node and merge content if present + let (existing_summary, existing_created_at) = + match store::read_node(config, namespace, hour_id)? { + Some(existing) => (Some(existing.summary), Some(existing.created_at)), + None => (None, None), + }; + + let to_summarize = if let Some(ref prev) = existing_summary { + format!("{prev}\n\n---\n\n{combined}") + } else { + combined + }; + + let hour_summary = summarize_to_limit( + provider, + &to_summarize, + NodeLevel::Hour.max_tokens(), + "hour", + hour_id, + ) + .await + .context("summarize hour leaf")?; + + let now = Utc::now(); + let hour_node = TreeNode { + node_id: hour_id.clone(), + namespace: namespace.to_string(), + level: NodeLevel::Hour, + parent_id: derive_parent_id(hour_id), + summary: hour_summary.clone(), + token_count: estimate_tokens(&hour_summary), + child_count: 0, + created_at: existing_created_at.unwrap_or(now), + updated_at: now, + metadata: None, + }; + store::write_node(config, &hour_node)?; + + publish_global(DomainEvent::TreeSummarizerHourCompleted { + namespace: namespace.to_string(), + node_id: hour_id.clone(), + token_count: hour_node.token_count, + }); + + tracing::debug!( + "[tree_summarizer] hour leaf {} created ({} tokens)", + hour_id, + hour_node.token_count + ); + + // Derive propagation path for this hour + let (_, day_id, month_id, year_id, root_id) = derive_node_ids_from_hour_id(hour_id); + all_propagation_ids.push((day_id, NodeLevel::Day)); + all_propagation_ids.push((month_id, NodeLevel::Month)); + all_propagation_ids.push((year_id, NodeLevel::Year)); + all_propagation_ids.push((root_id, NodeLevel::Root)); + + last_hour_node = Some(hour_node); + } + + // Deduplicate and propagate in bottom-up order (days, months, years, root) + let mut seen = std::collections::HashSet::new(); + for level in [ + NodeLevel::Day, + NodeLevel::Month, + NodeLevel::Year, + NodeLevel::Root, + ] { + for (node_id, node_level) in &all_propagation_ids { + if *node_level == level && seen.insert(node_id.clone()) { + propagate_node(config, provider, namespace, node_id, level) + .await + .with_context(|| format!("propagate {node_id}"))?; + } + } + } + + // All hour leaves are durably written and propagation is complete. + // Now it's safe to delete the buffer entries. + store::buffer_delete(config, namespace, &buffer_filenames) + .context("delete buffer entries after successful summarization")?; + + Ok(last_hour_node) +} + +/// Rebuild the entire tree from hour leaves upward. +/// Deletes all non-leaf nodes and re-summarizes. +/// Preserves buffered data that hasn't been summarized yet. +pub async fn rebuild_tree( + config: &Config, + provider: &dyn Provider, + namespace: &str, +) -> Result { + tracing::debug!("[tree_summarizer] rebuilding tree for namespace '{namespace}'"); + + let status = store::get_tree_status(config, namespace)?; + if status.total_nodes == 0 { + return Ok(status); + } + + // Collect all hour leaves first + let base = store::tree_dir(config, namespace); + let mut hour_leaves: Vec = Vec::new(); + collect_hour_leaves_recursive(&base, namespace, "", &mut hour_leaves)?; + + if hour_leaves.is_empty() { + tracing::debug!("[tree_summarizer] no hour leaves found, nothing to rebuild"); + return store::get_tree_status(config, namespace); + } + + // Preserve the buffer directory by moving it to a sibling path *outside* + // the tree directory, so delete_tree() does not destroy it. + let buffer_path = store::buffer_dir(config, namespace); + let tree_base = store::tree_dir(config, namespace); + // Place backup next to the tree dir (e.g. .../tree_buffer_backup) + let buffer_backup = tree_base + .parent() + .unwrap_or(&tree_base) + .join("tree_buffer_backup"); + let buffer_existed = buffer_path.exists(); + if buffer_existed { + if buffer_backup.exists() { + std::fs::remove_dir_all(&buffer_backup)?; + } + std::fs::rename(&buffer_path, &buffer_backup).context("backup buffer before rebuild")?; + tracing::debug!("[tree_summarizer] backed up buffer directory outside tree"); + } + + // Delete and recreate the tree directory + store::delete_tree(config, namespace)?; + + // Restore the buffer directory back inside the tree + if buffer_existed && buffer_backup.exists() { + let restored_buffer = store::buffer_dir(config, namespace); + if let Some(parent) = restored_buffer.parent() { + std::fs::create_dir_all(parent)?; + } + std::fs::rename(&buffer_backup, &restored_buffer) + .context("restore buffer after rebuild")?; + tracing::debug!("[tree_summarizer] restored buffer directory"); + } + + // Re-write all hour leaves + for leaf in &hour_leaves { + store::write_node(config, leaf)?; + } + + // Collect unique ancestor IDs at each level, ordered bottom-up + let mut day_ids = std::collections::BTreeSet::new(); + let mut month_ids = std::collections::BTreeSet::new(); + let mut year_ids = std::collections::BTreeSet::new(); + + for leaf in &hour_leaves { + if let Some(day) = derive_parent_id(&leaf.node_id) { + day_ids.insert(day.clone()); + if let Some(month) = derive_parent_id(&day) { + month_ids.insert(month.clone()); + if let Some(year) = derive_parent_id(&month) { + year_ids.insert(year); + } + } + } + } + + // Propagate bottom-up: days, then months, then years, then root + for day_id in &day_ids { + propagate_node(config, provider, namespace, day_id, NodeLevel::Day).await?; + } + for month_id in &month_ids { + propagate_node(config, provider, namespace, month_id, NodeLevel::Month).await?; + } + for year_id in &year_ids { + propagate_node(config, provider, namespace, year_id, NodeLevel::Year).await?; + } + propagate_node(config, provider, namespace, "root", NodeLevel::Root).await?; + + let final_status = store::get_tree_status(config, namespace)?; + + publish_global(DomainEvent::TreeSummarizerRebuildCompleted { + namespace: namespace.to_string(), + total_nodes: final_status.total_nodes, + }); + + tracing::debug!( + "[tree_summarizer] rebuild complete for '{}': {} nodes", + namespace, + final_status.total_nodes + ); + Ok(final_status) +} + +// ── Internal ─────────────────────────────────────────────────────────── + +/// Re-summarize a single non-leaf node from its children. +async fn propagate_node( + config: &Config, + provider: &dyn Provider, + namespace: &str, + node_id: &str, + level: NodeLevel, +) -> Result<()> { + let children = store::read_children(config, namespace, node_id)?; + if children.is_empty() { + tracing::debug!( + "[tree_summarizer] node {} has no children, skipping propagation", + node_id + ); + return Ok(()); + } + + let child_count = children.len() as u32; + let combined: String = children + .iter() + .map(|c| format!("## {} ({})\n\n{}", c.node_id, c.level.as_str(), c.summary)) + .collect::>() + .join("\n\n---\n\n"); + + let combined_tokens = estimate_tokens(&combined); + let max_tokens = level.max_tokens(); + + let summary = if combined_tokens <= max_tokens { + // Fits within budget — use the combined text directly + tracing::debug!( + "[tree_summarizer] node {} combined children ({} tokens) fits within {} token budget, no LLM needed", + node_id, + combined_tokens, + max_tokens + ); + combined + } else { + // Exceeds budget — summarize with LLM + tracing::debug!( + "[tree_summarizer] node {} combined children ({} tokens) exceeds {} token budget, summarizing", + node_id, + combined_tokens, + max_tokens + ); + summarize_to_limit(provider, &combined, max_tokens, level.as_str(), node_id).await? + }; + + let now = Utc::now(); + let existing = store::read_node(config, namespace, node_id)?; + let created_at = existing.map(|n| n.created_at).unwrap_or(now); + + let node = TreeNode { + node_id: node_id.to_string(), + namespace: namespace.to_string(), + level, + parent_id: derive_parent_id(node_id), + summary: summary.clone(), + token_count: estimate_tokens(&summary), + child_count, + created_at, + updated_at: now, + metadata: None, + }; + store::write_node(config, &node)?; + + publish_global(DomainEvent::TreeSummarizerPropagated { + namespace: namespace.to_string(), + node_id: node_id.to_string(), + level: level.as_str().to_string(), + token_count: node.token_count, + }); + + tracing::debug!( + "[tree_summarizer] propagated node {} (level={}, tokens={}, children={})", + node_id, + level.as_str(), + node.token_count, + child_count + ); + Ok(()) +} + +/// Summarize text to fit within a token limit using the LLM provider. +/// Enforces a hard character limit on the response to prevent runaway output. +async fn summarize_to_limit( + provider: &dyn Provider, + content: &str, + max_tokens: u32, + level_name: &str, + node_id: &str, +) -> Result { + let max_chars = (max_tokens as usize) * 4; + let system_prompt = format!( + "You are a hierarchical summarizer. Compress the following content into a concise \ + summary that preserves the most important information.\n\n\ + Rules:\n\ + - The summary MUST be under {max_tokens} tokens (roughly {max_chars} characters).\n\ + - Focus on key events, decisions, facts, patterns, and actionable insights.\n\ + - Preserve names, dates, numbers, and specific details when important.\n\ + - Use clear, dense prose — no filler.\n\n\ + Context: You are summarizing at the {level_name} level for node '{node_id}'.", + ); + + let response = provider + .chat_with_system( + Some(&system_prompt), + content, + SUMMARIZATION_MODEL, + SUMMARIZATION_TEMP, + ) + .await + .with_context(|| { + format!("LLM summarization failed for node {node_id} (level={level_name})") + })?; + + // Enforce hard character limit on LLM response (use the stricter of the two limits) + let char_limit = max_chars.min(MAX_SUMMARY_CHARS); + let response = if response.len() > char_limit { + tracing::warn!( + "[tree_summarizer] LLM response for node {} (level={}) was {} chars, truncating to {} chars", + node_id, + level_name, + response.len(), + char_limit + ); + // Truncate at a char boundary + let truncated = &response[..response.floor_char_boundary(char_limit)]; + truncated.to_string() + } else { + response + }; + + tracing::debug!( + "[tree_summarizer] LLM summarized {} chars -> {} chars for node {} (level={})", + content.len(), + response.len(), + node_id, + level_name + ); + + Ok(response) +} + +/// Group buffer entries by their hour based on filename timestamps. +/// +/// Buffer filenames are `{timestamp_millis}_{uuid}.md`. We extract the timestamp +/// and derive the hour ID for each entry. +fn group_by_hour(entries: &[(String, String)]) -> BTreeMap> { + let mut groups: BTreeMap> = BTreeMap::new(); + + for (filename, content) in entries { + let hour_id = hour_id_from_buffer_filename(filename).unwrap_or_else(|| { + // Fallback: use current time if filename can't be parsed + let now = Utc::now(); + let (hour, _, _, _, _) = derive_node_ids(&now); + hour + }); + groups.entry(hour_id).or_default().push(content.clone()); + } + + groups +} + +/// Extract the hour node ID from a buffer filename like `1711972800000_abc12345.md`. +fn hour_id_from_buffer_filename(filename: &str) -> Option { + let ts_str = filename.split('_').next()?; + let millis: i64 = ts_str.parse().ok()?; + let dt = DateTime::from_timestamp_millis(millis)?; + let (hour_id, _, _, _, _) = derive_node_ids(&dt); + Some(hour_id) +} + +/// Derive propagation IDs from an hour node_id string like "2024/03/15/14". +fn derive_node_ids_from_hour_id(hour_id: &str) -> (String, String, String, String, String) { + let parts: Vec<&str> = hour_id.split('/').collect(); + if parts.len() == 4 { + let year = parts[0].to_string(); + let month = format!("{}/{}", parts[0], parts[1]); + let day = format!("{}/{}/{}", parts[0], parts[1], parts[2]); + (hour_id.to_string(), day, month, year, "root".to_string()) + } else { + // Fallback + ( + hour_id.to_string(), + "unknown".to_string(), + "unknown".to_string(), + "unknown".to_string(), + "root".to_string(), + ) + } +} + +/// Recursively collect all hour leaf nodes from the tree directory. +fn collect_hour_leaves_recursive( + dir: &std::path::Path, + namespace: &str, + prefix: &str, + leaves: &mut Vec, +) -> Result<()> { + if !dir.exists() { + return Ok(()); + } + for entry in std::fs::read_dir(dir)? { + let entry = entry?; + let name = entry.file_name().to_string_lossy().to_string(); + let ft = entry.file_type()?; + + if ft.is_dir() { + if name == "buffer" || name == "buffer_backup" { + continue; + } + let child_prefix = if prefix.is_empty() { + name.clone() + } else { + format!("{prefix}/{name}") + }; + collect_hour_leaves_recursive(&entry.path(), namespace, &child_prefix, leaves)?; + } else if ft.is_file() && name.ends_with(".md") && name != "summary.md" && name != "root.md" + { + let hour_part = name.trim_end_matches(".md"); + let node_id = if prefix.is_empty() { + hour_part.to_string() + } else { + format!("{prefix}/{hour_part}") + }; + let level = level_from_node_id(&node_id); + if level == NodeLevel::Hour { + let raw = std::fs::read_to_string(entry.path())?; + let node = crate::openhuman::tree_summarizer::store::parse_node_markdown_pub( + &raw, namespace, &node_id, + ) + .with_context(|| format!("failed to parse hour leaf '{node_id}'"))?; + leaves.push(node); + } + } + } + Ok(()) +} + +// ── Hourly background loop ───────────────────────────────────────────── + +/// Start a background task that runs the summarization job every hour. +/// +/// This should be called once at application startup. The task runs +/// indefinitely, sleeping until the next hour boundary. +pub async fn run_hourly_loop(config: Config, provider: Box) { + tracing::debug!("[tree_summarizer] hourly loop started"); + + loop { + // Sleep until the next hour boundary + let now = Utc::now(); + let next_hour = { + let base = now + .date_naive() + .and_hms_opt(now.hour(), 0, 0) + .unwrap_or(now.naive_utc()); + let next = base + chrono::Duration::hours(1); + DateTime::::from_naive_utc_and_offset(next, Utc) + }; + let sleep_duration = (next_hour - now) + .to_std() + .unwrap_or(std::time::Duration::from_secs(3600)); + + tracing::debug!( + "[tree_summarizer] sleeping {:.0}s until next hour boundary", + sleep_duration.as_secs_f64() + ); + tokio::time::sleep(sleep_duration).await; + + // Run summarization for all namespaces that have buffered data + let ts = Utc::now(); + let namespaces = discover_active_namespaces(&config); + for ns in &namespaces { + match run_summarization(&config, provider.as_ref(), ns, ts).await { + Ok(Some(node)) => { + tracing::debug!( + "[tree_summarizer] hourly job completed for '{}': node {} ({} tokens)", + ns, + node.node_id, + node.token_count + ); + } + Ok(None) => { + tracing::debug!( + "[tree_summarizer] hourly job skipped for '{}' (no buffered data)", + ns + ); + } + Err(e) => { + tracing::error!("[tree_summarizer] hourly job failed for '{}': {:#}", ns, e); + } + } + } + } +} + +/// Discover namespaces that have pending buffer data by scanning the +/// `memory/namespaces/*/tree/buffer/` directories. +fn discover_active_namespaces(config: &Config) -> Vec { + let namespaces_dir = config.workspace_dir.join("memory").join("namespaces"); + + if !namespaces_dir.exists() { + return vec![]; + } + + let mut active = Vec::new(); + if let Ok(entries) = std::fs::read_dir(&namespaces_dir) { + for entry in entries.flatten() { + let name = entry.file_name().to_string_lossy().to_string(); + let buffer_dir = entry.path().join("tree").join("buffer"); + if buffer_dir.exists() { + // Check if buffer has any .md files + if let Ok(buffer_entries) = std::fs::read_dir(&buffer_dir) { + let has_entries = buffer_entries + .flatten() + .any(|e| e.path().extension().map(|ext| ext == "md").unwrap_or(false)); + if has_entries { + active.push(name); + } + } + } + } + } + active +} diff --git a/src/openhuman/tree_summarizer/mod.rs b/src/openhuman/tree_summarizer/mod.rs new file mode 100644 index 000000000..f9215777a --- /dev/null +++ b/src/openhuman/tree_summarizer/mod.rs @@ -0,0 +1,21 @@ +//! Hierarchical time-based summary tree. +//! +//! Organizes summaries as a tree: root → year → month → day → hour (leaf). +//! Each hour, a background job drains buffered raw content, summarizes it into +//! the hour leaf, and propagates updated summaries upward through the tree. +//! Stored as markdown files in `memory/namespaces/{ns}/tree/`. + +pub mod bus; +pub mod engine; +pub mod ops; +pub mod store; +pub mod types; + +mod schemas; + +pub use ops as rpc; +pub use schemas::{ + all_controller_schemas as all_tree_summarizer_controller_schemas, + all_registered_controllers as all_tree_summarizer_registered_controllers, +}; +pub use types::*; diff --git a/src/openhuman/tree_summarizer/ops.rs b/src/openhuman/tree_summarizer/ops.rs new file mode 100644 index 000000000..8a0a56a79 --- /dev/null +++ b/src/openhuman/tree_summarizer/ops.rs @@ -0,0 +1,157 @@ +//! RPC operation wrappers for the tree summarizer. + +use chrono::{DateTime, Utc}; +use serde_json::{json, Value}; + +use crate::openhuman::config::Config; +use crate::openhuman::providers; +use crate::openhuman::tree_summarizer::{engine, store, types::*}; +use crate::rpc::RpcOutcome; + +/// Append raw content to the ingestion buffer. +pub async fn tree_summarizer_ingest( + config: &Config, + namespace: &str, + content: &str, + timestamp: Option>, + metadata: Option<&Value>, +) -> Result, String> { + store::validate_namespace(namespace)?; + if content.trim().is_empty() { + return Err("content must not be empty".to_string()); + } + + let ts = timestamp.unwrap_or_else(Utc::now); + let path = store::buffer_write(config, namespace.trim(), content, &ts, metadata) + .map_err(|e| format!("buffer write failed: {e}"))?; + + Ok(RpcOutcome::single_log( + json!({ + "buffered": true, + "namespace": namespace.trim(), + "timestamp": ts.to_rfc3339(), + "tokens": estimate_tokens(content), + "path": path.display().to_string(), + "has_metadata": metadata.is_some(), + }), + format!("content buffered for namespace '{}'", namespace.trim()), + )) +} + +/// Trigger the summarization job for a namespace (drain buffer + summarize + propagate). +pub async fn tree_summarizer_run( + config: &Config, + namespace: &str, +) -> Result, String> { + store::validate_namespace(namespace)?; + + let provider = create_provider(config)?; + let ts = Utc::now(); + + match engine::run_summarization(config, provider.as_ref(), namespace.trim(), ts).await { + Ok(Some(node)) => Ok(RpcOutcome::single_log( + serde_json::to_value(&node).map_err(|e| e.to_string())?, + format!( + "summarization completed for '{}': node {} ({} tokens)", + namespace.trim(), + node.node_id, + node.token_count + ), + )), + Ok(None) => Ok(RpcOutcome::single_log( + json!({ "skipped": true, "reason": "no buffered data" }), + format!( + "summarization skipped for '{}': no buffered data", + namespace.trim() + ), + )), + Err(e) => Err(format!("summarization failed: {e:#}")), + } +} + +/// Query the tree at a specific node or level. +pub async fn tree_summarizer_query( + config: &Config, + namespace: &str, + node_id: Option<&str>, +) -> Result, String> { + store::validate_namespace(namespace)?; + + let target_id = node_id.unwrap_or("root"); + store::validate_node_id(target_id)?; + + let node = store::read_node(config, namespace.trim(), target_id) + .map_err(|e| format!("read node: {e}"))? + .ok_or_else(|| { + format!( + "node '{}' not found in namespace '{}'", + target_id, + namespace.trim() + ) + })?; + + let children = store::read_children(config, namespace.trim(), target_id) + .map_err(|e| format!("read children: {e}"))?; + + let result = QueryResult { node, children }; + Ok(RpcOutcome::single_log( + serde_json::to_value(&result).map_err(|e| e.to_string())?, + format!( + "queried node '{}' in namespace '{}'", + target_id, + namespace.trim() + ), + )) +} + +/// Get tree status/metadata for a namespace. +pub async fn tree_summarizer_status( + config: &Config, + namespace: &str, +) -> Result, String> { + store::validate_namespace(namespace)?; + + let status = + store::get_tree_status(config, namespace.trim()).map_err(|e| format!("get status: {e}"))?; + + Ok(RpcOutcome::single_log( + serde_json::to_value(&status).map_err(|e| e.to_string())?, + format!("tree status for namespace '{}'", namespace.trim()), + )) +} + +/// Rebuild the entire tree from hour leaves (background task). +pub async fn tree_summarizer_rebuild( + config: &Config, + namespace: &str, +) -> Result, String> { + store::validate_namespace(namespace)?; + + let provider = create_provider(config)?; + + let status = engine::rebuild_tree(config, provider.as_ref(), namespace.trim()) + .await + .map_err(|e| format!("rebuild failed: {e:#}"))?; + + Ok(RpcOutcome::single_log( + serde_json::to_value(&status).map_err(|e| e.to_string())?, + format!( + "tree rebuilt for '{}': {} nodes", + namespace.trim(), + status.total_nodes + ), + )) +} + +// ── Helper ───────────────────────────────────────────────────────────── + +fn create_provider( + config: &Config, +) -> Result, String> { + providers::create_resilient_provider( + config.api_key.as_deref(), + config.api_url.as_deref(), + &config.reliability, + ) + .map_err(|e| format!("failed to create provider: {e}")) +} diff --git a/src/openhuman/tree_summarizer/schemas.rs b/src/openhuman/tree_summarizer/schemas.rs new file mode 100644 index 000000000..3de27a5ca --- /dev/null +++ b/src/openhuman/tree_summarizer/schemas.rs @@ -0,0 +1,291 @@ +//! Controller schemas and RPC handler wiring for `tree_summarizer`. + +use serde::de::DeserializeOwned; +use serde_json::{Map, Value}; + +use crate::core::all::{ControllerFuture, RegisteredController}; +use crate::core::{ControllerSchema, FieldSchema, TypeSchema}; +use crate::openhuman::config::rpc as config_rpc; +use crate::rpc::RpcOutcome; + +pub fn all_controller_schemas() -> Vec { + vec![ + schemas("ingest"), + schemas("run"), + schemas("query"), + schemas("status"), + schemas("rebuild"), + ] +} + +pub fn all_registered_controllers() -> Vec { + vec![ + RegisteredController { + schema: schemas("ingest"), + handler: handle_ingest, + }, + RegisteredController { + schema: schemas("run"), + handler: handle_run, + }, + RegisteredController { + schema: schemas("query"), + handler: handle_query, + }, + RegisteredController { + schema: schemas("status"), + handler: handle_status, + }, + RegisteredController { + schema: schemas("rebuild"), + handler: handle_rebuild, + }, + ] +} + +fn namespace_input(comment: &'static str) -> FieldSchema { + FieldSchema { + name: "namespace", + ty: TypeSchema::String, + comment, + required: true, + } +} + +pub fn schemas(function: &str) -> ControllerSchema { + match function { + "ingest" => ControllerSchema { + namespace: "tree_summarizer", + function: "ingest", + description: "Append raw content to the tree summarizer ingestion buffer.", + inputs: vec![ + namespace_input("Namespace (scope) for the summary tree."), + FieldSchema { + name: "content", + ty: TypeSchema::String, + comment: "Raw content to buffer for summarization.", + required: true, + }, + FieldSchema { + name: "timestamp", + ty: TypeSchema::Option(Box::new(TypeSchema::String)), + comment: "Optional RFC3339 timestamp; defaults to now.", + required: false, + }, + FieldSchema { + name: "metadata", + ty: TypeSchema::Option(Box::new(TypeSchema::Json)), + comment: "Optional metadata JSON.", + required: false, + }, + ], + outputs: vec![FieldSchema { + name: "result", + ty: TypeSchema::Json, + comment: "Confirmation of buffered content.", + required: true, + }], + }, + "run" => ControllerSchema { + namespace: "tree_summarizer", + function: "run", + description: + "Trigger the summarization job: drain buffer, create hour leaf, propagate upward.", + inputs: vec![namespace_input( + "Namespace to run the summarization job for.", + )], + outputs: vec![FieldSchema { + name: "result", + ty: TypeSchema::Json, + comment: "Hour leaf node or skip status.", + required: true, + }], + }, + "query" => ControllerSchema { + namespace: "tree_summarizer", + function: "query", + description: "Read a tree node and its direct children.", + inputs: vec![ + namespace_input("Namespace of the summary tree."), + FieldSchema { + name: "node_id", + ty: TypeSchema::Option(Box::new(TypeSchema::String)), + comment: "Node ID to query; defaults to 'root'.", + required: false, + }, + ], + outputs: vec![FieldSchema { + name: "result", + ty: TypeSchema::Json, + comment: "The node and its children.", + required: true, + }], + }, + "status" => ControllerSchema { + namespace: "tree_summarizer", + function: "status", + description: "Get tree metadata: node count, depth, date range.", + inputs: vec![namespace_input("Namespace of the summary tree.")], + outputs: vec![FieldSchema { + name: "result", + ty: TypeSchema::Json, + comment: "Tree status metadata.", + required: true, + }], + }, + "rebuild" => ControllerSchema { + namespace: "tree_summarizer", + function: "rebuild", + description: + "Rebuild the entire summary tree from hour leaves upward (re-summarizes all levels).", + inputs: vec![namespace_input("Namespace to rebuild.")], + outputs: vec![FieldSchema { + name: "result", + ty: TypeSchema::Json, + comment: "Tree status after rebuild.", + required: true, + }], + }, + _other => ControllerSchema { + namespace: "tree_summarizer", + function: "unknown", + description: "Unknown tree_summarizer controller function.", + inputs: vec![FieldSchema { + name: "function", + ty: TypeSchema::String, + comment: "Unknown function requested for schema lookup.", + required: true, + }], + outputs: vec![FieldSchema { + name: "error", + ty: TypeSchema::String, + comment: "Lookup error details.", + required: true, + }], + }, + } +} + +// ── Handlers ─────────────────────────────────────────────────────────── + +fn handle_ingest(params: Map) -> ControllerFuture { + Box::pin(async move { + let config = config_rpc::load_config_with_timeout().await?; + let namespace = read_required::(¶ms, "namespace")?; + let content = read_required::(¶ms, "content")?; + let timestamp = read_optional_timestamp(¶ms, "timestamp")?; + let metadata = read_optional::(¶ms, "metadata")?; + to_json( + crate::openhuman::tree_summarizer::rpc::tree_summarizer_ingest( + &config, + &namespace, + &content, + timestamp, + metadata.as_ref(), + ) + .await?, + ) + }) +} + +fn handle_run(params: Map) -> ControllerFuture { + Box::pin(async move { + let config = config_rpc::load_config_with_timeout().await?; + let namespace = read_required::(¶ms, "namespace")?; + to_json( + crate::openhuman::tree_summarizer::rpc::tree_summarizer_run(&config, &namespace) + .await?, + ) + }) +} + +fn handle_query(params: Map) -> ControllerFuture { + Box::pin(async move { + let config = config_rpc::load_config_with_timeout().await?; + let namespace = read_required::(¶ms, "namespace")?; + let node_id = read_optional::(¶ms, "node_id")?; + to_json( + crate::openhuman::tree_summarizer::rpc::tree_summarizer_query( + &config, + &namespace, + node_id.as_deref(), + ) + .await?, + ) + }) +} + +fn handle_status(params: Map) -> ControllerFuture { + Box::pin(async move { + let config = config_rpc::load_config_with_timeout().await?; + let namespace = read_required::(¶ms, "namespace")?; + to_json( + crate::openhuman::tree_summarizer::rpc::tree_summarizer_status(&config, &namespace) + .await?, + ) + }) +} + +fn handle_rebuild(params: Map) -> ControllerFuture { + Box::pin(async move { + let config = config_rpc::load_config_with_timeout().await?; + let namespace = read_required::(¶ms, "namespace")?; + to_json( + crate::openhuman::tree_summarizer::rpc::tree_summarizer_rebuild(&config, &namespace) + .await?, + ) + }) +} + +// ── Param helpers ────────────────────────────────────────────────────── + +fn read_required(params: &Map, key: &str) -> Result { + let value = params + .get(key) + .cloned() + .ok_or_else(|| format!("missing required param '{key}'"))?; + serde_json::from_value(value).map_err(|e| format!("invalid '{key}': {e}")) +} + +fn read_optional( + params: &Map, + key: &str, +) -> Result, String> { + match params.get(key) { + None | Some(Value::Null) => Ok(None), + Some(v) => serde_json::from_value(v.clone()) + .map(Some) + .map_err(|e| format!("invalid '{key}': {e}")), + } +} + +fn read_optional_timestamp( + params: &Map, + key: &str, +) -> Result>, String> { + match params.get(key) { + None | Some(Value::Null) => Ok(None), + Some(Value::String(s)) => chrono::DateTime::parse_from_rfc3339(s) + .map(|dt| Some(dt.with_timezone(&chrono::Utc))) + .map_err(|e| format!("invalid '{key}': {e}")), + Some(other) => Err(format!( + "invalid '{key}': expected string, got {}", + type_name(other) + )), + } +} + +fn to_json(outcome: RpcOutcome) -> Result { + outcome.into_cli_compatible_json() +} + +fn type_name(value: &Value) -> &'static str { + match value { + Value::Null => "null", + Value::Bool(_) => "bool", + Value::Number(_) => "number", + Value::String(_) => "string", + Value::Array(_) => "array", + Value::Object(_) => "object", + } +} diff --git a/src/openhuman/tree_summarizer/store.rs b/src/openhuman/tree_summarizer/store.rs new file mode 100644 index 000000000..233e88e67 --- /dev/null +++ b/src/openhuman/tree_summarizer/store.rs @@ -0,0 +1,953 @@ +//! Markdown file-based persistence for the summary tree. +//! +//! Each tree node is stored as a markdown file with YAML frontmatter in the +//! memory namespaces directory: +//! `{workspace}/memory/namespaces/{namespace}/tree/` +//! +//! The folder hierarchy mirrors the time hierarchy: +//! root.md, 2024/summary.md, 2024/03/summary.md, 2024/03/15/summary.md, 2024/03/15/14.md + +use anyhow::{Context, Result}; +use chrono::{DateTime, TimeZone, Utc}; +use serde_json::Value; +use std::path::{Path, PathBuf}; + +use crate::openhuman::config::Config; +use crate::openhuman::tree_summarizer::types::{ + derive_parent_id, estimate_tokens, level_from_node_id, node_id_to_path, NodeLevel, TreeNode, + TreeStatus, +}; + +// ── Path helpers ─────────────────────────────────────────────────────── + +/// Base tree directory for a namespace. +pub fn tree_dir(config: &Config, namespace: &str) -> PathBuf { + config + .workspace_dir + .join("memory") + .join("namespaces") + .join(sanitize(namespace)) + .join("tree") +} + +/// Buffer directory where raw ingested content is staged before summarization. +pub fn buffer_dir(config: &Config, namespace: &str) -> PathBuf { + tree_dir(config, namespace).join("buffer") +} + +/// Absolute file path for a given node. +pub fn node_file_path(config: &Config, namespace: &str, node_id: &str) -> PathBuf { + tree_dir(config, namespace).join(node_id_to_path(node_id)) +} + +/// Sanitize a namespace string for use as a directory name. +/// Rejects namespaces containing path-traversal or reserved characters. +fn sanitize(namespace: &str) -> String { + let trimmed = namespace.trim(); + // Replace characters that are unsafe for directory names + trimmed + .replace(['/', '\\', ':', '*', '?', '"', '<', '>', '|', '.'], "_") + .replace("__", "_") +} + +/// Validate a namespace string, returning an error for empty or dangerous input. +pub fn validate_namespace(namespace: &str) -> Result<(), String> { + let trimmed = namespace.trim(); + if trimmed.is_empty() { + return Err("namespace must not be empty".to_string()); + } + if trimmed.contains("..") { + return Err("namespace must not contain '..'".to_string()); + } + if trimmed.starts_with('/') || trimmed.starts_with('\\') { + return Err("namespace must not start with a path separator".to_string()); + } + Ok(()) +} + +/// Validate a node_id against the allowed canonical formats. +/// Accepts: "root", "YYYY", "YYYY/MM", "YYYY/MM/DD", "YYYY/MM/DD/HH". +/// Rejects path traversal, empty segments, and non-numeric components. +pub fn validate_node_id(node_id: &str) -> Result<(), String> { + if node_id == "root" { + return Ok(()); + } + + // Reject path traversal and dangerous characters + if node_id.contains("..") || node_id.starts_with('/') || node_id.ends_with('/') { + return Err(format!( + "invalid node_id '{node_id}': contains path traversal or leading/trailing slashes" + )); + } + + let parts: Vec<&str> = node_id.split('/').collect(); + if parts.is_empty() || parts.len() > 4 { + return Err(format!( + "invalid node_id '{node_id}': expected 1-4 segments (YYYY[/MM[/DD[/HH]]])" + )); + } + + // All parts must be non-empty numeric strings + for (i, part) in parts.iter().enumerate() { + if part.is_empty() { + return Err(format!( + "invalid node_id '{node_id}': empty segment at position {i}" + )); + } + if !part.chars().all(|c| c.is_ascii_digit()) { + return Err(format!( + "invalid node_id '{node_id}': non-numeric segment '{part}' at position {i}" + )); + } + } + + // Basic range validation + if parts.len() >= 2 { + let month: u32 = parts[1].parse().unwrap_or(0); + if !(1..=12).contains(&month) { + return Err(format!( + "invalid node_id '{node_id}': month {month} out of range 1-12" + )); + } + } + if parts.len() >= 3 { + let day: u32 = parts[2].parse().unwrap_or(0); + if !(1..=31).contains(&day) { + return Err(format!( + "invalid node_id '{node_id}': day {day} out of range 1-31" + )); + } + } + if parts.len() >= 4 { + let hour: u32 = parts[3].parse().unwrap_or(99); + if hour > 23 { + return Err(format!( + "invalid node_id '{node_id}': hour {hour} out of range 0-23" + )); + } + } + + Ok(()) +} + +// ── Write ────────────────────────────────────────────────────────────── + +/// Write a tree node to disk as a markdown file with YAML frontmatter. +pub fn write_node(config: &Config, node: &TreeNode) -> Result<()> { + let path = node_file_path(config, &node.namespace, &node.node_id); + if let Some(parent) = path.parent() { + std::fs::create_dir_all(parent) + .with_context(|| format!("create dirs for {}", parent.display()))?; + } + + let metadata_line = match &node.metadata { + Some(m) => format!("metadata: {m}\n"), + None => String::new(), + }; + + let frontmatter = format!( + "---\n\ + node_id: \"{}\"\n\ + namespace: \"{}\"\n\ + level: {}\n\ + parent_id: {}\n\ + token_count: {}\n\ + child_count: {}\n\ + created_at: {}\n\ + updated_at: {}\n\ + {}\ + ---\n\n", + node.node_id, + node.namespace, + node.level.as_str(), + match &node.parent_id { + Some(pid) => format!("\"{pid}\""), + None => "~".to_string(), + }, + node.token_count, + node.child_count, + node.created_at.to_rfc3339(), + node.updated_at.to_rfc3339(), + metadata_line, + ); + + let content = format!("{frontmatter}{}\n", node.summary); + std::fs::write(&path, content) + .with_context(|| format!("write tree node {}", path.display()))?; + + tracing::debug!( + "[tree_summarizer] wrote node {} (level={}, tokens={}) -> {}", + node.node_id, + node.level.as_str(), + node.token_count, + path.display() + ); + Ok(()) +} + +// ── Read ─────────────────────────────────────────────────────────────── + +/// Read a single tree node from its markdown file. Returns `None` if the file +/// does not exist. +pub fn read_node(config: &Config, namespace: &str, node_id: &str) -> Result> { + let path = node_file_path(config, namespace, node_id); + if !path.exists() { + return Ok(None); + } + let raw = std::fs::read_to_string(&path) + .with_context(|| format!("read tree node {}", path.display()))?; + parse_node_markdown(&raw, namespace, node_id).map(Some) +} + +/// Read all direct children of a node. +pub fn read_children(config: &Config, namespace: &str, parent_id: &str) -> Result> { + let parent_level = level_from_node_id(parent_id); + let base = tree_dir(config, namespace); + + match parent_level { + NodeLevel::Root => read_subdirectory_summaries(&base, namespace, ""), + NodeLevel::Year | NodeLevel::Month => { + read_subdirectory_summaries(&base, namespace, parent_id) + } + NodeLevel::Day => read_hour_leaves(&base, namespace, parent_id), + NodeLevel::Hour => Ok(vec![]), // leaves have no children + } +} + +/// Walk up from a node to the root, returning all ancestors (excluding the node itself). +pub fn read_ancestors(config: &Config, namespace: &str, node_id: &str) -> Result> { + let mut ancestors = Vec::new(); + let mut current = derive_parent_id(node_id); + while let Some(pid) = current { + if let Some(node) = read_node(config, namespace, &pid)? { + ancestors.push(node); + } + current = derive_parent_id(&pid); + } + Ok(ancestors) +} + +/// Recursively count all `.md` files in the tree directory. +pub fn count_nodes(config: &Config, namespace: &str) -> Result { + let base = tree_dir(config, namespace); + if !base.exists() { + return Ok(0); + } + count_md_files(&base) +} + +/// Scan the tree to produce a status summary. +pub fn get_tree_status(config: &Config, namespace: &str) -> Result { + let base = tree_dir(config, namespace); + let total_nodes = if base.exists() { + count_md_files(&base)? + } else { + 0 + }; + + // Determine depth by checking which levels exist. + let mut depth = 0u32; + let root_path = base.join("root.md"); + if root_path.exists() { + depth = 1; + } + + // Scan for years/months/days/hours to figure out actual depth and date range. + let mut oldest: Option> = None; + let mut newest: Option> = None; + + if base.exists() { + for entry in std::fs::read_dir(&base).into_iter().flatten().flatten() { + let name = entry.file_name().to_string_lossy().to_string(); + if entry.file_type().map(|t| t.is_dir()).unwrap_or(false) && name.len() == 4 { + if depth < 2 { + depth = 2; + } + // Scan months, days, hours inside + let year_dir = entry.path(); + for month_entry in std::fs::read_dir(&year_dir).into_iter().flatten().flatten() { + if month_entry.file_type().map(|t| t.is_dir()).unwrap_or(false) { + if depth < 3 { + depth = 3; + } + let month_dir = month_entry.path(); + for day_entry in std::fs::read_dir(&month_dir) + .into_iter() + .flatten() + .flatten() + { + if day_entry.file_type().map(|t| t.is_dir()).unwrap_or(false) { + if depth < 4 { + depth = 4; + } + // Check for hour .md files + let day_dir = day_entry.path(); + for hour_entry in + std::fs::read_dir(&day_dir).into_iter().flatten().flatten() + { + let hname = + hour_entry.file_name().to_string_lossy().to_string(); + if hname.ends_with(".md") && hname != "summary.md" { + if depth < 5 { + depth = 5; + } + // Try to parse timestamp from path + if let Some(ts) = timestamp_from_hour_path( + &name, + &month_entry.file_name().to_string_lossy().to_string(), + &day_entry.file_name().to_string_lossy().to_string(), + &hname, + ) { + match &oldest { + None => oldest = Some(ts), + Some(o) if ts < *o => oldest = Some(ts), + _ => {} + } + match &newest { + None => newest = Some(ts), + Some(n) if ts > *n => newest = Some(ts), + _ => {} + } + } + } + } + } + } + } + } + } + } + } + + Ok(TreeStatus { + namespace: namespace.to_string(), + total_nodes, + depth, + oldest_entry: oldest, + newest_entry: newest, + last_run_at: None, // filled by caller if needed + }) +} + +/// Remove the entire tree directory for a namespace. +pub fn delete_tree(config: &Config, namespace: &str) -> Result { + let base = tree_dir(config, namespace); + if !base.exists() { + return Ok(0); + } + let count = count_md_files(&base)?; + std::fs::remove_dir_all(&base).with_context(|| format!("delete tree at {}", base.display()))?; + tracing::debug!( + "[tree_summarizer] deleted tree for namespace '{}' ({} nodes)", + namespace, + count + ); + Ok(count) +} + +// ── Buffer operations ────────────────────────────────────────────────── + +/// Append raw content to the ingestion buffer as a timestamped file. +/// Optionally includes metadata as a JSON object stored alongside the content. +pub fn buffer_write( + config: &Config, + namespace: &str, + content: &str, + ts: &DateTime, + metadata: Option<&Value>, +) -> Result { + let dir = buffer_dir(config, namespace); + std::fs::create_dir_all(&dir) + .with_context(|| format!("create buffer dir {}", dir.display()))?; + + let filename = format!( + "{}_{}.md", + ts.timestamp_millis(), + &uuid::Uuid::new_v4().to_string()[..8] + ); + let path = dir.join(&filename); + + // If metadata is provided, write it as a YAML frontmatter block + let file_content = if let Some(meta) = metadata { + let meta_str = serde_json::to_string(meta).unwrap_or_default(); + format!("---\nmetadata: {meta_str}\n---\n\n{content}") + } else { + content.to_string() + }; + + std::fs::write(&path, file_content) + .with_context(|| format!("write buffer entry {}", path.display()))?; + + tracing::debug!( + "[tree_summarizer] buffered {} bytes for namespace '{}' -> {}", + content.len(), + namespace, + filename + ); + Ok(path) +} + +/// Read all buffered entries non-destructively, returning `(filename, content)` pairs +/// sorted by filename (chronological). Files remain on disk until explicitly deleted +/// via [`buffer_delete`]. +pub fn buffer_read(config: &Config, namespace: &str) -> Result> { + let dir = buffer_dir(config, namespace); + if !dir.exists() { + return Ok(vec![]); + } + + let mut entries: Vec<(String, PathBuf)> = Vec::new(); + for entry in std::fs::read_dir(&dir)? { + let entry = entry?; + let path = entry.path(); + if path.extension().map(|e| e == "md").unwrap_or(false) { + let name = entry.file_name().to_string_lossy().to_string(); + entries.push((name, path)); + } + } + + entries.sort_by(|a, b| a.0.cmp(&b.0)); + + let mut contents = Vec::with_capacity(entries.len()); + for (name, path) in &entries { + let raw = std::fs::read_to_string(path) + .with_context(|| format!("read buffer entry {}", path.display()))?; + // Strip metadata frontmatter if present, pass raw content + let text = strip_buffer_frontmatter(&raw); + contents.push((name.clone(), text)); + } + + tracing::debug!( + "[tree_summarizer] read {} buffer entries for namespace '{}'", + contents.len(), + namespace + ); + Ok(contents) +} + +/// Delete specific buffer entries by filename after they have been successfully +/// processed and durably written as hour leaves. +pub fn buffer_delete(config: &Config, namespace: &str, filenames: &[String]) -> Result<()> { + let dir = buffer_dir(config, namespace); + for name in filenames { + let path = dir.join(name); + if path.exists() { + std::fs::remove_file(&path).with_context(|| { + format!( + "failed to remove buffer entry '{}' at {}", + name, + path.display() + ) + })?; + } + } + tracing::debug!( + "[tree_summarizer] deleted {} buffer entries for namespace '{}'", + filenames.len(), + namespace + ); + Ok(()) +} + +/// Read and drain all buffered entries. Convenience wrapper that calls +/// [`buffer_read`] then [`buffer_delete`]. Use the split API when you need +/// to defer deletion until after durable writes complete. +pub fn buffer_drain(config: &Config, namespace: &str) -> Result> { + let entries = buffer_read(config, namespace)?; + if entries.is_empty() { + return Ok(entries); + } + let filenames: Vec = entries.iter().map(|(name, _)| name.clone()).collect(); + buffer_delete(config, namespace, &filenames)?; + tracing::debug!( + "[tree_summarizer] drained {} buffer entries for namespace '{}'", + entries.len(), + namespace + ); + Ok(entries) +} + +/// Strip the optional metadata frontmatter from a buffer entry, +/// returning only the content body. +fn strip_buffer_frontmatter(raw: &str) -> String { + let trimmed = raw.trim_start(); + if !trimmed.starts_with("---") { + return raw.to_string(); + } + let after_open = &trimmed[3..]; + if let Some(close_pos) = after_open.find("\n---") { + let body_start = close_pos + 4; + after_open[body_start..] + .trim_start_matches('\n') + .to_string() + } else { + raw.to_string() + } +} + +// ── Internal helpers ─────────────────────────────────────────────────── + +/// Read summary.md files from subdirectories of a given parent path. +fn read_subdirectory_summaries( + base: &Path, + namespace: &str, + parent_id: &str, +) -> Result> { + let scan_dir = if parent_id.is_empty() { + base.to_path_buf() + } else { + base.join(parent_id) + }; + if !scan_dir.exists() { + return Ok(vec![]); + } + + let mut children = Vec::new(); + for entry in std::fs::read_dir(&scan_dir)? { + let entry = entry?; + if !entry.file_type().map(|t| t.is_dir()).unwrap_or(false) { + continue; + } + let child_name = entry.file_name().to_string_lossy().to_string(); + // Skip non-numeric directories and the buffer directory + if child_name == "buffer" + || child_name == "buffer_backup" + || child_name.chars().any(|c| !c.is_ascii_digit()) + { + continue; + } + let child_id = if parent_id.is_empty() { + child_name + } else { + format!("{parent_id}/{child_name}") + }; + let summary_path = entry.path().join("summary.md"); + if summary_path.exists() { + let raw = std::fs::read_to_string(&summary_path)?; + if let Ok(node) = parse_node_markdown(&raw, namespace, &child_id) { + children.push(node); + } + } + } + + children.sort_by(|a, b| a.node_id.cmp(&b.node_id)); + Ok(children) +} + +/// Read hour leaf .md files (excluding summary.md) from a day directory. +fn read_hour_leaves(base: &Path, namespace: &str, day_id: &str) -> Result> { + let day_dir = base.join(day_id); + if !day_dir.exists() { + return Ok(vec![]); + } + + let mut leaves = Vec::new(); + for entry in std::fs::read_dir(&day_dir)? { + let entry = entry?; + let name = entry.file_name().to_string_lossy().to_string(); + if !name.ends_with(".md") || name == "summary.md" { + continue; + } + let hour_part = name.trim_end_matches(".md"); + let node_id = format!("{day_id}/{hour_part}"); + let raw = std::fs::read_to_string(entry.path())?; + if let Ok(node) = parse_node_markdown(&raw, namespace, &node_id) { + leaves.push(node); + } + } + + leaves.sort_by(|a, b| a.node_id.cmp(&b.node_id)); + Ok(leaves) +} + +/// Public entry point for parsing a markdown node (used by engine rebuild). +pub fn parse_node_markdown_pub(raw: &str, namespace: &str, node_id: &str) -> Result { + parse_node_markdown(raw, namespace, node_id) +} + +/// Parse a markdown file with YAML frontmatter into a `TreeNode`. +fn parse_node_markdown(raw: &str, namespace: &str, node_id: &str) -> Result { + let (frontmatter, body_raw) = split_frontmatter(raw); + let body = body_raw.trim_end().to_string(); + + let level = frontmatter + .get("level") + .and_then(|v| NodeLevel::from_str_label(v)) + .unwrap_or_else(|| level_from_node_id(node_id)); + + let parent_id = frontmatter + .get("parent_id") + .and_then(|v| { + let trimmed = v.trim().trim_matches('"'); + if trimmed == "~" || trimmed.is_empty() { + None + } else { + Some(trimmed.to_string()) + } + }) + .or_else(|| derive_parent_id(node_id)); + + let token_count = frontmatter + .get("token_count") + .and_then(|v| v.parse::().ok()) + .unwrap_or_else(|| estimate_tokens(&body)); + + let child_count = frontmatter + .get("child_count") + .and_then(|v| v.parse::().ok()) + .unwrap_or(0); + + let created_at = frontmatter + .get("created_at") + .and_then(|v| DateTime::parse_from_rfc3339(v).ok()) + .map(|dt| dt.with_timezone(&Utc)) + .unwrap_or_else(Utc::now); + + let updated_at = frontmatter + .get("updated_at") + .and_then(|v| DateTime::parse_from_rfc3339(v).ok()) + .map(|dt| dt.with_timezone(&Utc)) + .unwrap_or_else(Utc::now); + + let metadata = frontmatter.get("metadata").map(|v| v.to_string()); + + Ok(TreeNode { + node_id: node_id.to_string(), + namespace: namespace.to_string(), + level, + parent_id, + summary: body, + token_count, + child_count, + created_at, + updated_at, + metadata, + }) +} + +/// Split markdown into (frontmatter key-value map, body text). +fn split_frontmatter(raw: &str) -> (std::collections::HashMap, String) { + let mut map = std::collections::HashMap::new(); + let trimmed = raw.trim_start(); + + if !trimmed.starts_with("---") { + return (map, raw.to_string()); + } + + // Find the closing --- + let after_open = &trimmed[3..]; + if let Some(close_pos) = after_open.find("\n---") { + let fm_block = &after_open[..close_pos]; + let body_start = close_pos + 4; // skip "\n---" + let body = after_open[body_start..] + .trim_start_matches('\n') + .to_string(); + + for line in fm_block.lines() { + let line = line.trim(); + if line.is_empty() { + continue; + } + if let Some(colon_pos) = line.find(':') { + let key = line[..colon_pos].trim().to_string(); + let value = line[colon_pos + 1..].trim().trim_matches('"').to_string(); + map.insert(key, value); + } + } + + (map, body) + } else { + (map, raw.to_string()) + } +} + +fn count_md_files(dir: &Path) -> Result { + let mut count = 0u64; + for entry in std::fs::read_dir(dir)? { + let entry = entry?; + let ft = entry.file_type()?; + if ft.is_dir() { + let name = entry.file_name().to_string_lossy().to_string(); + if name == "buffer" || name == "buffer_backup" { + continue; // skip buffer directories + } + count += count_md_files(&entry.path())?; + } else if ft.is_file() { + if entry.path().extension().map(|e| e == "md").unwrap_or(false) { + count += 1; + } + } + } + Ok(count) +} + +fn timestamp_from_hour_path( + year: &str, + month: &str, + day: &str, + hour_file: &str, +) -> Option> { + let hour = hour_file.trim_end_matches(".md"); + let y: i32 = year.parse().ok()?; + let m: u32 = month.parse().ok()?; + let d: u32 = day.parse().ok()?; + let h: u32 = hour.parse().ok()?; + chrono::Utc.with_ymd_and_hms(y, m, d, h, 0, 0).single() +} + +// ── Tests ────────────────────────────────────────────────────────────── + +#[cfg(test)] +mod tests { + use super::*; + use tempfile::TempDir; + + fn test_config(tmp: &TempDir) -> Config { + let config = Config { + workspace_dir: tmp.path().join("workspace"), + config_path: tmp.path().join("config.toml"), + ..Config::default() + }; + std::fs::create_dir_all(&config.workspace_dir).unwrap(); + config + } + + fn make_node(namespace: &str, node_id: &str, summary: &str) -> TreeNode { + let level = level_from_node_id(node_id); + TreeNode { + node_id: node_id.to_string(), + namespace: namespace.to_string(), + level, + parent_id: derive_parent_id(node_id), + summary: summary.to_string(), + token_count: estimate_tokens(summary), + child_count: 0, + created_at: Utc::now(), + updated_at: Utc::now(), + metadata: None, + } + } + + #[test] + fn write_and_read_node_roundtrip() { + let tmp = TempDir::new().unwrap(); + let config = test_config(&tmp); + let ns = "test-ns"; + + let node = make_node(ns, "root", "All-time summary of events."); + write_node(&config, &node).unwrap(); + + let read_back = read_node(&config, ns, "root").unwrap().unwrap(); + assert_eq!(read_back.node_id, "root"); + assert_eq!(read_back.level, NodeLevel::Root); + assert_eq!(read_back.summary, "All-time summary of events."); + assert!(read_back.parent_id.is_none()); + } + + #[test] + fn write_and_read_hour_leaf() { + let tmp = TempDir::new().unwrap(); + let config = test_config(&tmp); + let ns = "test-ns"; + + let node = make_node(ns, "2024/03/15/14", "Hour 14 summary."); + write_node(&config, &node).unwrap(); + + let read_back = read_node(&config, ns, "2024/03/15/14").unwrap().unwrap(); + assert_eq!(read_back.level, NodeLevel::Hour); + assert_eq!(read_back.parent_id.as_deref(), Some("2024/03/15")); + assert_eq!(read_back.summary, "Hour 14 summary."); + } + + #[test] + fn read_children_of_day() { + let tmp = TempDir::new().unwrap(); + let config = test_config(&tmp); + let ns = "test-ns"; + + // Write some hour leaves + for hour in [10, 11, 14] { + let node = make_node( + ns, + &format!("2024/03/15/{hour:02}"), + &format!("Hour {hour}."), + ); + write_node(&config, &node).unwrap(); + } + // Write the day summary (should not appear as a child) + let day = make_node(ns, "2024/03/15", "Day summary."); + write_node(&config, &day).unwrap(); + + let children = read_children(&config, ns, "2024/03/15").unwrap(); + assert_eq!(children.len(), 3); + assert_eq!(children[0].node_id, "2024/03/15/10"); + assert_eq!(children[1].node_id, "2024/03/15/11"); + assert_eq!(children[2].node_id, "2024/03/15/14"); + } + + #[test] + fn read_children_of_root() { + let tmp = TempDir::new().unwrap(); + let config = test_config(&tmp); + let ns = "test-ns"; + + for year in ["2023", "2024"] { + let node = make_node(ns, year, &format!("Year {year} summary.")); + write_node(&config, &node).unwrap(); + } + + let children = read_children(&config, ns, "root").unwrap(); + assert_eq!(children.len(), 2); + assert_eq!(children[0].node_id, "2023"); + assert_eq!(children[1].node_id, "2024"); + } + + #[test] + fn read_node_missing_returns_none() { + let tmp = TempDir::new().unwrap(); + let config = test_config(&tmp); + assert!(read_node(&config, "ns", "root").unwrap().is_none()); + } + + #[test] + fn count_nodes_and_status() { + let tmp = TempDir::new().unwrap(); + let config = test_config(&tmp); + let ns = "test-ns"; + + write_node(&config, &make_node(ns, "root", "root")).unwrap(); + write_node(&config, &make_node(ns, "2024", "year")).unwrap(); + write_node(&config, &make_node(ns, "2024/03", "month")).unwrap(); + write_node(&config, &make_node(ns, "2024/03/15", "day")).unwrap(); + write_node(&config, &make_node(ns, "2024/03/15/14", "hour")).unwrap(); + + assert_eq!(count_nodes(&config, ns).unwrap(), 5); + + let status = get_tree_status(&config, ns).unwrap(); + assert_eq!(status.total_nodes, 5); + assert_eq!(status.depth, 5); + } + + #[test] + fn delete_tree_removes_all() { + let tmp = TempDir::new().unwrap(); + let config = test_config(&tmp); + let ns = "test-ns"; + + write_node(&config, &make_node(ns, "root", "root")).unwrap(); + write_node(&config, &make_node(ns, "2024/03/15/14", "hour")).unwrap(); + + let deleted = delete_tree(&config, ns).unwrap(); + assert!(deleted >= 2); + assert_eq!(count_nodes(&config, ns).unwrap(), 0); + } + + #[test] + fn buffer_write_and_drain() { + let tmp = TempDir::new().unwrap(); + let config = test_config(&tmp); + let ns = "test-ns"; + let ts1 = Utc.with_ymd_and_hms(2024, 3, 15, 10, 0, 0).unwrap(); + let ts2 = Utc.with_ymd_and_hms(2024, 3, 15, 11, 0, 0).unwrap(); + + buffer_write(&config, ns, "entry one", &ts1, None).unwrap(); + buffer_write(&config, ns, "entry two", &ts2, None).unwrap(); + + let drained = buffer_drain(&config, ns).unwrap(); + assert_eq!(drained.len(), 2); + // Sorted by filename (timestamp prefix), so ts1 < ts2 + assert_eq!(drained[0].1, "entry one"); + assert_eq!(drained[1].1, "entry two"); + + // Buffer should be empty now + let again = buffer_drain(&config, ns).unwrap(); + assert!(again.is_empty()); + } + + #[test] + fn buffer_write_with_metadata() { + let tmp = TempDir::new().unwrap(); + let config = test_config(&tmp); + let ns = "test-ns"; + let now = Utc::now(); + + let meta = serde_json::json!({"source": "test", "priority": 1}); + buffer_write(&config, ns, "entry with meta", &now, Some(&meta)).unwrap(); + + let drained = buffer_drain(&config, ns).unwrap(); + assert_eq!(drained.len(), 1); + // Content should be stripped of frontmatter + assert_eq!(drained[0].1, "entry with meta"); + } + + #[test] + fn ancestors_walk_to_root() { + let tmp = TempDir::new().unwrap(); + let config = test_config(&tmp); + let ns = "test-ns"; + + write_node(&config, &make_node(ns, "root", "root")).unwrap(); + write_node(&config, &make_node(ns, "2024", "year")).unwrap(); + write_node(&config, &make_node(ns, "2024/03", "month")).unwrap(); + write_node(&config, &make_node(ns, "2024/03/15", "day")).unwrap(); + + let ancestors = read_ancestors(&config, ns, "2024/03/15/14").unwrap(); + let ids: Vec<&str> = ancestors.iter().map(|n| n.node_id.as_str()).collect(); + assert_eq!(ids, vec!["2024/03/15", "2024/03", "2024", "root"]); + } + + #[test] + fn frontmatter_parsing() { + let raw = "---\nnode_id: \"root\"\nlevel: root\ntoken_count: 42\n---\n\nHello world."; + let (fm, body) = split_frontmatter(raw); + assert_eq!(fm.get("level").unwrap(), "root"); + assert_eq!(fm.get("token_count").unwrap(), "42"); + assert_eq!(body, "Hello world."); + } + + #[test] + fn validate_node_id_accepts_valid() { + assert!(validate_node_id("root").is_ok()); + assert!(validate_node_id("2024").is_ok()); + assert!(validate_node_id("2024/03").is_ok()); + assert!(validate_node_id("2024/03/15").is_ok()); + assert!(validate_node_id("2024/03/15/14").is_ok()); + } + + #[test] + fn validate_node_id_rejects_traversal() { + assert!(validate_node_id("..").is_err()); + assert!(validate_node_id("../etc").is_err()); + assert!(validate_node_id("2024/../etc").is_err()); + assert!(validate_node_id("/2024").is_err()); + assert!(validate_node_id("2024/").is_err()); + } + + #[test] + fn validate_node_id_rejects_non_numeric() { + assert!(validate_node_id("abc").is_err()); + assert!(validate_node_id("2024/abc").is_err()); + assert!(validate_node_id("2024/03/15/foo").is_err()); + } + + #[test] + fn validate_node_id_rejects_out_of_range() { + assert!(validate_node_id("2024/13").is_err()); // month 13 + assert!(validate_node_id("2024/03/32").is_err()); // day 32 + assert!(validate_node_id("2024/03/15/24").is_err()); // hour 24 + } + + #[test] + fn validate_namespace_rejects_dangerous() { + assert!(validate_namespace("").is_err()); + assert!(validate_namespace(" ").is_err()); + assert!(validate_namespace("../etc").is_err()); + assert!(validate_namespace("/absolute").is_err()); + } + + #[test] + fn validate_namespace_accepts_valid() { + assert!(validate_namespace("my-namespace").is_ok()); + assert!(validate_namespace("skill:gmail:user@example.com").is_ok()); + } +} diff --git a/src/openhuman/tree_summarizer/types.rs b/src/openhuman/tree_summarizer/types.rs new file mode 100644 index 000000000..04eb46cff --- /dev/null +++ b/src/openhuman/tree_summarizer/types.rs @@ -0,0 +1,294 @@ +//! Domain types for the tree summarizer. + +use chrono::{DateTime, Datelike, Timelike, Utc}; +use serde::{Deserialize, Serialize}; +use std::path::PathBuf; + +// ── Node level ───────────────────────────────────────────────────────── + +/// Hierarchical level of a tree node. +#[derive(Debug, Clone, Copy, PartialEq, Eq, Hash, Serialize, Deserialize)] +#[serde(rename_all = "snake_case")] +pub enum NodeLevel { + Root, + Year, + Month, + Day, + Hour, +} + +impl NodeLevel { + /// Maximum number of tokens allowed at this level. + pub fn max_tokens(&self) -> u32 { + match self { + Self::Hour => 1_000, + Self::Day => 2_000, + Self::Month => 4_000, + Self::Year => 8_000, + Self::Root => 20_000, + } + } + + /// The level above this one in the hierarchy (`None` for root). + pub fn parent_level(&self) -> Option { + match self { + Self::Hour => Some(Self::Day), + Self::Day => Some(Self::Month), + Self::Month => Some(Self::Year), + Self::Year => Some(Self::Root), + Self::Root => None, + } + } + + /// True only for the leaf level (hour). + pub fn is_leaf(&self) -> bool { + matches!(self, Self::Hour) + } + + /// Parse a level string from YAML frontmatter. + pub fn from_str_label(s: &str) -> Option { + match s.trim().to_ascii_lowercase().as_str() { + "root" => Some(Self::Root), + "year" => Some(Self::Year), + "month" => Some(Self::Month), + "day" => Some(Self::Day), + "hour" => Some(Self::Hour), + _ => None, + } + } + + /// Label for display / frontmatter. + pub fn as_str(&self) -> &'static str { + match self { + Self::Root => "root", + Self::Year => "year", + Self::Month => "month", + Self::Day => "day", + Self::Hour => "hour", + } + } +} + +// ── Tree node ────────────────────────────────────────────────────────── + +/// A single node in the summary tree. +#[derive(Debug, Clone, Serialize, Deserialize)] +pub struct TreeNode { + pub node_id: String, + pub namespace: String, + pub level: NodeLevel, + pub parent_id: Option, + pub summary: String, + pub token_count: u32, + pub child_count: u32, + pub created_at: DateTime, + pub updated_at: DateTime, + #[serde(default, skip_serializing_if = "Option::is_none")] + pub metadata: Option, +} + +// ── Status ───────────────────────────────────────────────────────────── + +/// Metadata about an entire tree within a namespace. +#[derive(Debug, Clone, Serialize, Deserialize)] +pub struct TreeStatus { + pub namespace: String, + pub total_nodes: u64, + pub depth: u32, + pub oldest_entry: Option>, + pub newest_entry: Option>, + pub last_run_at: Option>, +} + +// ── Ingest request ───────────────────────────────────────────────────── + +/// Input for appending raw content to the ingestion buffer. +#[derive(Debug, Clone, Serialize, Deserialize)] +pub struct IngestRequest { + pub namespace: String, + pub content: String, + #[serde(default)] + pub timestamp: Option>, + #[serde(default, skip_serializing_if = "Option::is_none")] + pub metadata: Option, +} + +// ── Query result ─────────────────────────────────────────────────────── + +/// Result of a tree query at a specific node. +#[derive(Debug, Clone, Serialize, Deserialize)] +pub struct QueryResult { + pub node: TreeNode, + pub children: Vec, +} + +// ── Helpers ──────────────────────────────────────────────────────────── + +/// Rough token estimate: ~4 characters per token. +pub fn estimate_tokens(text: &str) -> u32 { + (text.len() as u32 + 3) / 4 +} + +/// Derive the parent node ID from a node ID. +/// +/// - `"2024/03/15/14"` → `Some("2024/03/15")` +/// - `"2024/03/15"` → `Some("2024/03")` +/// - `"2024/03"` → `Some("2024")` +/// - `"2024"` → `Some("root")` +/// - `"root"` → `None` +pub fn derive_parent_id(node_id: &str) -> Option { + if node_id == "root" { + return None; + } + match node_id.rfind('/') { + Some(pos) => Some(node_id[..pos].to_string()), + None => Some("root".to_string()), + } +} + +/// Determine the `NodeLevel` from a node ID string. +pub fn level_from_node_id(node_id: &str) -> NodeLevel { + if node_id == "root" { + return NodeLevel::Root; + } + match node_id.matches('/').count() { + 0 => NodeLevel::Year, // "2024" + 1 => NodeLevel::Month, // "2024/03" + 2 => NodeLevel::Day, // "2024/03/15" + _ => NodeLevel::Hour, // "2024/03/15/14" + } +} + +/// Derive all ancestor node IDs from a timestamp (hour through root). +/// +/// Returns `(hour_id, day_id, month_id, year_id, root_id)`. +pub fn derive_node_ids(ts: &DateTime) -> (String, String, String, String, String) { + let year = format!("{}", ts.year()); + let month = format!("{}/{:02}", ts.year(), ts.month()); + let day = format!("{}/{:02}/{:02}", ts.year(), ts.month(), ts.day()); + let hour = format!( + "{}/{:02}/{:02}/{:02}", + ts.year(), + ts.month(), + ts.day(), + ts.hour() + ); + (hour, day, month, year, "root".to_string()) +} + +/// Convert a node ID to a relative file path within the tree directory. +/// +/// - `"root"` → `root.md` +/// - `"2024"` → `2024/summary.md` +/// - `"2024/03"` → `2024/03/summary.md` +/// - `"2024/03/15"` → `2024/03/15/summary.md` +/// - `"2024/03/15/14"` → `2024/03/15/14.md` (hour leaf — file, not folder) +pub fn node_id_to_path(node_id: &str) -> PathBuf { + if node_id == "root" { + return PathBuf::from("root.md"); + } + let level = level_from_node_id(node_id); + if level.is_leaf() { + // Hour leaf: "2024/03/15/14" → "2024/03/15/14.md" + PathBuf::from(format!("{node_id}.md")) + } else { + // Non-leaf: "2024/03" → "2024/03/summary.md" + PathBuf::from(node_id).join("summary.md") + } +} + +// ── Tests ────────────────────────────────────────────────────────────── + +#[cfg(test)] +mod tests { + use super::*; + use chrono::TimeZone; + + #[test] + fn node_level_max_tokens() { + assert_eq!(NodeLevel::Hour.max_tokens(), 1_000); + assert_eq!(NodeLevel::Day.max_tokens(), 2_000); + assert_eq!(NodeLevel::Month.max_tokens(), 4_000); + assert_eq!(NodeLevel::Year.max_tokens(), 8_000); + assert_eq!(NodeLevel::Root.max_tokens(), 20_000); + } + + #[test] + fn node_level_parent_chain() { + assert_eq!(NodeLevel::Hour.parent_level(), Some(NodeLevel::Day)); + assert_eq!(NodeLevel::Day.parent_level(), Some(NodeLevel::Month)); + assert_eq!(NodeLevel::Month.parent_level(), Some(NodeLevel::Year)); + assert_eq!(NodeLevel::Year.parent_level(), Some(NodeLevel::Root)); + assert_eq!(NodeLevel::Root.parent_level(), None); + } + + #[test] + fn derive_parent_id_chain() { + assert_eq!(derive_parent_id("2024/03/15/14"), Some("2024/03/15".into())); + assert_eq!(derive_parent_id("2024/03/15"), Some("2024/03".into())); + assert_eq!(derive_parent_id("2024/03"), Some("2024".into())); + assert_eq!(derive_parent_id("2024"), Some("root".into())); + assert_eq!(derive_parent_id("root"), None); + } + + #[test] + fn level_from_node_id_all_levels() { + assert_eq!(level_from_node_id("root"), NodeLevel::Root); + assert_eq!(level_from_node_id("2024"), NodeLevel::Year); + assert_eq!(level_from_node_id("2024/03"), NodeLevel::Month); + assert_eq!(level_from_node_id("2024/03/15"), NodeLevel::Day); + assert_eq!(level_from_node_id("2024/03/15/14"), NodeLevel::Hour); + } + + #[test] + fn derive_node_ids_from_timestamp() { + let ts = Utc.with_ymd_and_hms(2024, 3, 15, 14, 30, 0).unwrap(); + let (hour, day, month, year, root) = derive_node_ids(&ts); + assert_eq!(hour, "2024/03/15/14"); + assert_eq!(day, "2024/03/15"); + assert_eq!(month, "2024/03"); + assert_eq!(year, "2024"); + assert_eq!(root, "root"); + } + + #[test] + fn node_id_to_path_mapping() { + assert_eq!(node_id_to_path("root"), PathBuf::from("root.md")); + assert_eq!(node_id_to_path("2024"), PathBuf::from("2024/summary.md")); + assert_eq!( + node_id_to_path("2024/03"), + PathBuf::from("2024/03/summary.md") + ); + assert_eq!( + node_id_to_path("2024/03/15"), + PathBuf::from("2024/03/15/summary.md") + ); + assert_eq!( + node_id_to_path("2024/03/15/14"), + PathBuf::from("2024/03/15/14.md") + ); + } + + #[test] + fn estimate_tokens_rough() { + assert_eq!(estimate_tokens(""), 0); + assert_eq!(estimate_tokens("abcd"), 1); + assert_eq!(estimate_tokens("abcdefgh"), 2); + // Roughly 4 chars per token + let text = "a".repeat(4000); + assert_eq!(estimate_tokens(&text), 1000); + } + + #[test] + fn node_level_roundtrip() { + for level in [ + NodeLevel::Root, + NodeLevel::Year, + NodeLevel::Month, + NodeLevel::Day, + NodeLevel::Hour, + ] { + assert_eq!(NodeLevel::from_str_label(level.as_str()), Some(level)); + } + } +}