mirror of
https://github.com/tinyhumansai/openhuman.git
synced 2026-07-27 21:08:00 +00:00
feat(tree-summarizer): hierarchical summary tree module with CLI (#423)
* feat(tree-summarizer): implement hierarchical summarization engine and event handling - Introduced a new `tree_summarizer` module to manage hierarchical time-based summaries, organizing data into a tree structure (root → year → month → day → hour). - Added functionality to ingest raw content, summarize it into hour leaves, and propagate summaries upward through the tree. - Implemented event handling for summarization completion and tree rebuild events, enhancing observability and modularity. - Created RPC operations for ingesting content, triggering summarization, querying the tree, and retrieving tree status. - Added comprehensive tests to ensure the reliability of the summarization process and event handling. This update significantly enhances the summarization capabilities of the system, allowing for efficient data organization and retrieval. * feat(tree-summarizer): add CLI support for tree summarization commands - Introduced a new `tree-summarizer` command to the CLI, allowing users to ingest content, run summarization jobs, query the summary tree, check status, and rebuild the tree. - Updated the CLI help documentation to include the new command and its subcommands. - Added a new module `tree_summarizer_cli` to encapsulate the tree summarization functionality. This enhancement improves the usability of the summarization features, providing a streamlined interface for managing hierarchical summaries directly from the command line. * style: apply cargo fmt to tree_summarizer module * feat(tree-summarizer): implement TreeSummarizerEventSubscriber for observability logging - Added a new `TreeSummarizerEventSubscriber` to log events related to tree summarization, enhancing observability. - Updated the `start_channels` function to register the new subscriber. - Refactored the `run_summarization` function to group buffered entries by hour and publish events upon completion of summarization. - Improved documentation and added tests for the new subscriber functionality. This update aims to provide better insights into the summarization process and facilitate future cross-module workflows. * refactor(tree_summarizer): streamline buffer backup and function signature - Simplified the buffer backup process by consolidating the rename operation with context handling for better error reporting. - Cleaned up the function signature of `derive_node_ids_from_hour_id` for improved readability. These changes enhance code clarity and maintainability within the tree summarization engine. * feat(tree_summarizer): enhance tree summarization with metadata support - Updated the `tree_summarizer_ingest` function to accept an optional metadata parameter, allowing users to include additional context during content ingestion. - Refactored related functions to validate and handle metadata, improving the overall robustness of the summarization process. - Adjusted the buffer write functionality to store metadata alongside content, enhancing the data structure for future retrieval and processing. These changes aim to enrich the summarization capabilities and provide more context for ingested content. * refactor(tree_summarizer): improve error message formatting in node ID validation - Enhanced the formatting of error messages in the `validate_node_id` function for better readability and consistency. - Adjusted the string formatting to use multi-line syntax, improving clarity in error reporting. - Minor formatting changes in the `strip_buffer_frontmatter` function to enhance code readability. These changes aim to improve the maintainability and clarity of error handling within the tree summarization module. * refactor(tree_summarizer): enhance buffer management and summarization process - Replaced the buffer draining mechanism with a non-destructive read approach, allowing for safer data handling during summarization. - Introduced a new `buffer_delete` function to explicitly manage the deletion of buffer entries after successful processing. - Updated the `run_summarization` function to reflect these changes, ensuring that buffer entries are only deleted after durable writes are confirmed. - Improved the backup process for the buffer directory during tree rebuilds, ensuring it is preserved outside the tree structure. These modifications aim to improve data integrity and clarity in the summarization workflow. * refactor(tree_summarizer): improve markdown parsing and timestamp handling - Updated the `parse_node_markdown` function to trim trailing whitespace from the body after splitting frontmatter, enhancing data cleanliness. - Modified test cases to use specific timestamps instead of the current time, ensuring consistent and predictable test results. - Adjusted assertions in tests to reflect the new timestamp-based ordering of entries. These changes aim to improve the robustness of markdown parsing and the reliability of test outcomes in the tree summarization module.
This commit is contained in:
Steven Enamakel
Generated
+1
-1
@@ -4,7 +4,7 @@ version = 4
|
||||
|
||||
[[package]]
|
||||
name = "OpenHuman"
|
||||
version = "0.51.17"
|
||||
version = "0.51.18"
|
||||
dependencies = [
|
||||
"env_logger",
|
||||
"log",
|
||||
|
||||
@@ -74,6 +74,8 @@ fn build_registered_controllers() -> Vec<RegisteredController> {
|
||||
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<ControllerSchema> {
|
||||
@@ -111,6 +113,7 @@ fn build_declared_controller_schemas() -> Vec<ControllerSchema> {
|
||||
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,
|
||||
}
|
||||
}
|
||||
|
||||
@@ -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<String, Vec<ControllerSchema>>) {
|
||||
println!(" openhuman call --method <name> [--params '<json>']");
|
||||
println!(" openhuman skills <subcommand> [options] (skill development runtime)");
|
||||
println!(" openhuman voice [--hotkey <combo>] [--mode <tap|push>] (voice dictation server)");
|
||||
println!(" openhuman tree-summarizer <subcommand> [options] (summary tree CLI)");
|
||||
println!(" openhuman <namespace> <function> [--param value ...]\n");
|
||||
println!("Available namespaces:");
|
||||
for namespace in grouped.keys() {
|
||||
|
||||
@@ -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.
|
||||
|
||||
@@ -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 <namespace> [--content <text> | --file <path>] [-v]
|
||||
//! openhuman tree-summarizer run <namespace> [-v]
|
||||
//! openhuman tree-summarizer query <namespace> [<node_id>] [-v]
|
||||
//! openhuman tree-summarizer status <namespace> [-v]
|
||||
//! openhuman tree-summarizer rebuild <namespace> [-v]
|
||||
|
||||
use anyhow::Result;
|
||||
|
||||
/// Entry point for `openhuman tree-summarizer <subcommand>`.
|
||||
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<String>,
|
||||
file: Option<String>,
|
||||
node_id: Option<String>,
|
||||
}
|
||||
|
||||
fn parse_opts(args: &[String]) -> Result<(CliOpts, Vec<String>)> {
|
||||
let mut verbose = false;
|
||||
let mut content: Option<String> = None;
|
||||
let mut file: Option<String> = None;
|
||||
let mut node_id: Option<String> = 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 <namespace> --content <text>` or `--file <path>`
|
||||
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 <namespace> [--content <text>] [--file <path>] [-v]");
|
||||
println!();
|
||||
println!("Append content to the summarization buffer for a namespace.");
|
||||
println!();
|
||||
println!(" <namespace> Target namespace for the summary tree");
|
||||
println!(" --content, -c <text> Raw text content to ingest");
|
||||
println!(" --file, -f <path> 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 <namespace>`
|
||||
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 <namespace> [-v]");
|
||||
println!();
|
||||
println!("Trigger the summarization job for a namespace.");
|
||||
println!("Drains the buffer, creates the hour leaf, and propagates upward.");
|
||||
println!();
|
||||
println!(" <namespace> 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 <namespace> [<node_id>]`
|
||||
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 <namespace> [<node_id>] [--node-id <id>] [-v]"
|
||||
);
|
||||
println!();
|
||||
println!("Read a summary tree node and its direct children.");
|
||||
println!();
|
||||
println!(" <namespace> Target namespace");
|
||||
println!(" <node_id> 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 <namespace>`
|
||||
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 <namespace> [-v]");
|
||||
println!();
|
||||
println!("Show tree metadata: node count, depth, date range.");
|
||||
println!();
|
||||
println!(" <namespace> 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 <namespace>`
|
||||
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 <namespace> [-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!(" <namespace> 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::Runtime> {
|
||||
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<crate::openhuman::config::Config> {
|
||||
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 <namespace> [--content <text>] [--file <path>] [-v]"
|
||||
);
|
||||
println!(" openhuman tree-summarizer run <namespace> [-v]");
|
||||
println!(" openhuman tree-summarizer query <namespace> [<node_id>] [-v]");
|
||||
println!(" openhuman tree-summarizer status <namespace> [-v]");
|
||||
println!(" openhuman tree-summarizer rebuild <namespace> [-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");
|
||||
}
|
||||
@@ -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());
|
||||
|
||||
|
||||
@@ -142,6 +142,23 @@ pub enum DomainEvent {
|
||||
error: Option<String>,
|
||||
},
|
||||
|
||||
// ── 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 {
|
||||
|
||||
@@ -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;
|
||||
|
||||
@@ -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
|
||||
}
|
||||
}
|
||||
@@ -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<Utc>,
|
||||
) -> Result<Option<TreeNode>> {
|
||||
// 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<String> = 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<TreeNode> = 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<TreeStatus> {
|
||||
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<TreeNode> = 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::<Vec<_>>()
|
||||
.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<String> {
|
||||
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<String, Vec<String>> {
|
||||
let mut groups: BTreeMap<String, Vec<String>> = 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<String> {
|
||||
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<TreeNode>,
|
||||
) -> 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<dyn Provider>) {
|
||||
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::<Utc>::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<String> {
|
||||
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
|
||||
}
|
||||
@@ -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::*;
|
||||
@@ -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<DateTime<Utc>>,
|
||||
metadata: Option<&Value>,
|
||||
) -> Result<RpcOutcome<Value>, 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<RpcOutcome<Value>, 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<RpcOutcome<Value>, 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<RpcOutcome<Value>, 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<RpcOutcome<Value>, 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<Box<dyn crate::openhuman::providers::traits::Provider>, 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}"))
|
||||
}
|
||||
@@ -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<ControllerSchema> {
|
||||
vec![
|
||||
schemas("ingest"),
|
||||
schemas("run"),
|
||||
schemas("query"),
|
||||
schemas("status"),
|
||||
schemas("rebuild"),
|
||||
]
|
||||
}
|
||||
|
||||
pub fn all_registered_controllers() -> Vec<RegisteredController> {
|
||||
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<String, Value>) -> ControllerFuture {
|
||||
Box::pin(async move {
|
||||
let config = config_rpc::load_config_with_timeout().await?;
|
||||
let namespace = read_required::<String>(¶ms, "namespace")?;
|
||||
let content = read_required::<String>(¶ms, "content")?;
|
||||
let timestamp = read_optional_timestamp(¶ms, "timestamp")?;
|
||||
let metadata = read_optional::<Value>(¶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<String, Value>) -> ControllerFuture {
|
||||
Box::pin(async move {
|
||||
let config = config_rpc::load_config_with_timeout().await?;
|
||||
let namespace = read_required::<String>(¶ms, "namespace")?;
|
||||
to_json(
|
||||
crate::openhuman::tree_summarizer::rpc::tree_summarizer_run(&config, &namespace)
|
||||
.await?,
|
||||
)
|
||||
})
|
||||
}
|
||||
|
||||
fn handle_query(params: Map<String, Value>) -> ControllerFuture {
|
||||
Box::pin(async move {
|
||||
let config = config_rpc::load_config_with_timeout().await?;
|
||||
let namespace = read_required::<String>(¶ms, "namespace")?;
|
||||
let node_id = read_optional::<String>(¶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<String, Value>) -> ControllerFuture {
|
||||
Box::pin(async move {
|
||||
let config = config_rpc::load_config_with_timeout().await?;
|
||||
let namespace = read_required::<String>(¶ms, "namespace")?;
|
||||
to_json(
|
||||
crate::openhuman::tree_summarizer::rpc::tree_summarizer_status(&config, &namespace)
|
||||
.await?,
|
||||
)
|
||||
})
|
||||
}
|
||||
|
||||
fn handle_rebuild(params: Map<String, Value>) -> ControllerFuture {
|
||||
Box::pin(async move {
|
||||
let config = config_rpc::load_config_with_timeout().await?;
|
||||
let namespace = read_required::<String>(¶ms, "namespace")?;
|
||||
to_json(
|
||||
crate::openhuman::tree_summarizer::rpc::tree_summarizer_rebuild(&config, &namespace)
|
||||
.await?,
|
||||
)
|
||||
})
|
||||
}
|
||||
|
||||
// ── Param helpers ──────────────────────────────────────────────────────
|
||||
|
||||
fn read_required<T: DeserializeOwned>(params: &Map<String, Value>, key: &str) -> Result<T, String> {
|
||||
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<T: DeserializeOwned>(
|
||||
params: &Map<String, Value>,
|
||||
key: &str,
|
||||
) -> Result<Option<T>, 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<String, Value>,
|
||||
key: &str,
|
||||
) -> Result<Option<chrono::DateTime<chrono::Utc>>, 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<T: serde::Serialize>(outcome: RpcOutcome<T>) -> Result<Value, String> {
|
||||
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",
|
||||
}
|
||||
}
|
||||
@@ -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<Option<TreeNode>> {
|
||||
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<Vec<TreeNode>> {
|
||||
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<Vec<TreeNode>> {
|
||||
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<u64> {
|
||||
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<TreeStatus> {
|
||||
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<DateTime<Utc>> = None;
|
||||
let mut newest: Option<DateTime<Utc>> = 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<u64> {
|
||||
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<Utc>,
|
||||
metadata: Option<&Value>,
|
||||
) -> Result<PathBuf> {
|
||||
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<Vec<(String, String)>> {
|
||||
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<Vec<(String, String)>> {
|
||||
let entries = buffer_read(config, namespace)?;
|
||||
if entries.is_empty() {
|
||||
return Ok(entries);
|
||||
}
|
||||
let filenames: Vec<String> = 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<Vec<TreeNode>> {
|
||||
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<Vec<TreeNode>> {
|
||||
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<TreeNode> {
|
||||
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<TreeNode> {
|
||||
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::<u32>().ok())
|
||||
.unwrap_or_else(|| estimate_tokens(&body));
|
||||
|
||||
let child_count = frontmatter
|
||||
.get("child_count")
|
||||
.and_then(|v| v.parse::<u32>().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, String>, 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<u64> {
|
||||
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<DateTime<Utc>> {
|
||||
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());
|
||||
}
|
||||
}
|
||||
@@ -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<NodeLevel> {
|
||||
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<Self> {
|
||||
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<String>,
|
||||
pub summary: String,
|
||||
pub token_count: u32,
|
||||
pub child_count: u32,
|
||||
pub created_at: DateTime<Utc>,
|
||||
pub updated_at: DateTime<Utc>,
|
||||
#[serde(default, skip_serializing_if = "Option::is_none")]
|
||||
pub metadata: Option<String>,
|
||||
}
|
||||
|
||||
// ── 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<DateTime<Utc>>,
|
||||
pub newest_entry: Option<DateTime<Utc>>,
|
||||
pub last_run_at: Option<DateTime<Utc>>,
|
||||
}
|
||||
|
||||
// ── 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<DateTime<Utc>>,
|
||||
#[serde(default, skip_serializing_if = "Option::is_none")]
|
||||
pub metadata: Option<serde_json::Value>,
|
||||
}
|
||||
|
||||
// ── 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<TreeNode>,
|
||||
}
|
||||
|
||||
// ── 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<String> {
|
||||
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<Utc>) -> (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));
|
||||
}
|
||||
}
|
||||
}
|
||||
Reference in New Issue
Block a user