diff --git a/src/openhuman/tools/impl/memory/tree/ingest_document.rs b/src/openhuman/tools/impl/memory/tree/ingest_document.rs new file mode 100644 index 000000000..a8cb4e620 --- /dev/null +++ b/src/openhuman/tools/impl/memory/tree/ingest_document.rs @@ -0,0 +1,141 @@ +use crate::openhuman::config::rpc as config_rpc; +use crate::openhuman::memory::tree::canonicalize::document::DocumentInput; +use crate::openhuman::memory::tree::rpc; +use crate::openhuman::memory::tree::types::SourceKind; +use crate::openhuman::tools::traits::{Tool, ToolResult}; +use async_trait::async_trait; +use chrono::Utc; +use serde_json::json; + +pub struct MemoryTreeIngestDocumentTool; + +#[async_trait] +impl Tool for MemoryTreeIngestDocumentTool { + fn name(&self) -> &str { + "memory_tree_ingest_document" + } + + fn description(&self) -> &str { + "Ingest a document into the memory tree for future retrieval. \ + This is the write path into the knowledge index — use it after \ + fetching web content, extracting facts, or collecting data from \ + external sources. The ingested document will be chunked, embedded, \ + and available via query_global, query_source, and search_entities." + } + + fn parameters_schema(&self) -> serde_json::Value { + json!({ + "type": "object", + "properties": { + "title": { + "type": "string", + "description": "Document title (e.g. 'ROOT v6.36.12 Release Notes')." + }, + "body": { + "type": "string", + "description": "Document body in markdown or plain text." + }, + "source_id": { + "type": "string", + "description": "Stable source identifier (e.g. 'root_releases', 'github_root_changelog'). Re-ingesting with same source_id replaces old chunks." + }, + "provider": { + "type": "string", + "description": "Source provider name (e.g. 'github', 'web', 'root_docs'). Defaults to 'agent'." + }, + "source_ref": { + "type": "string", + "description": "Optional URL or pointer back to the original source." + }, + "owner": { + "type": "string", + "description": "Optional account/user this content belongs to. Used for owner-scoped queries and attribution. Defaults to empty (unowned/agent-global)." + } + }, + "required": ["title", "body", "source_id"] + }) + } + + async fn execute(&self, args: serde_json::Value) -> anyhow::Result { + log::debug!("[tool][memory_tree] ingest_document invoked"); + + let title = args + .get("title") + .and_then(|v| v.as_str()) + .ok_or_else(|| anyhow::anyhow!("ingest_document: missing required field `title`"))? + .to_string(); + let body = args + .get("body") + .and_then(|v| v.as_str()) + .ok_or_else(|| anyhow::anyhow!("ingest_document: missing required field `body`"))? + .to_string(); + let source_id = args + .get("source_id") + .and_then(|v| v.as_str()) + .ok_or_else(|| anyhow::anyhow!("ingest_document: missing required field `source_id`"))? + .trim() + .to_string(); + let provider = args + .get("provider") + .and_then(|v| v.as_str()) + .unwrap_or("agent") + .to_string(); + let source_ref = args + .get("source_ref") + .and_then(|v| v.as_str()) + .map(|s| s.to_string()); + let owner = args + .get("owner") + .and_then(|v| v.as_str()) + .unwrap_or("") + .to_string(); + + if title.trim().is_empty() || body.trim().is_empty() || source_id.is_empty() { + return Ok(ToolResult::error( + "ingest_document: title, body, and source_id must be non-empty".to_string(), + )); + } + + let cfg = config_rpc::load_config_with_timeout().await.map_err(|e| { + log::debug!("[tool][memory_tree] ingest_document config_load_failed err={e}"); + anyhow::anyhow!("ingest_document: load config failed: {e}") + })?; + + let doc = DocumentInput { + provider, + title: title.trim().to_string(), + body: body.trim().to_string(), + modified_at: Utc::now(), + source_ref, + }; + + let req = rpc::IngestRequest { + source_kind: SourceKind::Document, + source_id: source_id.clone(), + owner, + tags: vec!["agent_ingested".to_string()], + payload: serde_json::to_value(&doc).map_err(|e| { + log::debug!("[tool][memory_tree] ingest_document payload_serialize_failed err={e}"); + anyhow::anyhow!("ingest_document: failed to serialize payload: {e}") + })?, + }; + + let outcome = rpc::ingest_rpc(&cfg, req).await.map_err(|e| { + log::debug!( + "[tool][memory_tree] ingest_document rpc_failed source_id={source_id} err={e}" + ); + anyhow::anyhow!("ingest_document: ingestion failed: {e}") + })?; + + let n = outcome.value.chunks_written; + log::info!( + "[tool][memory_tree] ingest_document done source_id={} chunks={}", + source_id, + n + ); + Ok(ToolResult::success(format!( + "Ingested document \"{}\" as source_id={}. {} chunks created and indexed.", + title, source_id, n + ))) + } +} diff --git a/src/openhuman/tools/impl/memory/tree/mod.rs b/src/openhuman/tools/impl/memory/tree/mod.rs index c0b4c99a8..2030ae10c 100644 --- a/src/openhuman/tools/impl/memory/tree/mod.rs +++ b/src/openhuman/tools/impl/memory/tree/mod.rs @@ -9,6 +9,7 @@ mod drill_down; mod fetch_leaves; +mod ingest_document; mod query_global; mod query_source; mod query_topic; @@ -18,6 +19,7 @@ mod search_entities; // (e.g. tool registration in ops.rs). pub use drill_down::MemoryTreeDrillDownTool; pub use fetch_leaves::MemoryTreeFetchLeavesTool; +pub use ingest_document::MemoryTreeIngestDocumentTool; pub use query_global::MemoryTreeQueryGlobalTool; pub use query_source::MemoryTreeQuerySourceTool; pub use query_topic::MemoryTreeQueryTopicTool; @@ -46,7 +48,7 @@ impl Tool for MemoryTreeTool { `query_source` (filter by source type + time window), \ `query_global` (cross-source daily digest), \ `drill_down` (expand a coarse summary one level), \ - `fetch_leaves` (pull raw chunks for citation)." + `fetch_leaves` (pull raw chunks for citation), `ingest_document` (write a document into the tree for future retrieval)." } fn parameters_schema(&self) -> serde_json::Value { @@ -56,8 +58,8 @@ impl Tool for MemoryTreeTool { "mode": { "type": "string", "enum": ["search_entities", "query_topic", "query_source", - "query_global", "drill_down", "fetch_leaves"], - "description": "Which retrieval operation to run." + "query_global", "drill_down", "fetch_leaves", "ingest_document"], + "description": "Which operation to run (retrieval or write)." }, // search_entities params "query": { @@ -93,6 +95,27 @@ impl Tool for MemoryTreeTool { "description": "drill_down: how many levels to expand (default 1, max 3)." }, // fetch_leaves params + // ingest_document params + "title": { + "type": "string", + "description": "ingest_document: document title." + }, + "body": { + "type": "string", + "description": "ingest_document: document body (markdown or plain text)." + }, + "source_id": { + "type": "string", + "description": "ingest_document / query_source: stable source identifier. For ingest, re-ingesting same id replaces old chunks." + }, + "provider": { + "type": "string", + "description": "ingest_document: source provider (e.g. github, web, root_docs). Defaults to agent." + }, + "source_ref": { + "type": "string", + "description": "ingest_document: optional URL back to original source." + }, "chunk_ids": { "type": "array", "items": {"type": "string"}, @@ -121,9 +144,13 @@ impl Tool for MemoryTreeTool { "query_global" => MemoryTreeQueryGlobalTool.execute(args).await, "drill_down" => MemoryTreeDrillDownTool.execute(args).await, "fetch_leaves" => MemoryTreeFetchLeavesTool.execute(args).await, - other => Err(anyhow::anyhow!( - "memory_tree: unknown mode `{other}`. Valid: search_entities, query_topic, query_source, query_global, drill_down, fetch_leaves" - )), + "ingest_document" => MemoryTreeIngestDocumentTool.execute(args).await, + other => { + log::debug!("[tool][memory_tree] unknown_mode mode={other}"); + Err(anyhow::anyhow!( + "memory_tree: unknown mode `{other}`. Valid: search_entities, query_topic, query_source, query_global, drill_down, fetch_leaves, ingest_document" + )) + } } } } @@ -147,7 +174,7 @@ mod memory_tree_dispatcher_tests { } #[test] - fn memory_tree_schema_mode_enum_has_all_six_modes() { + fn memory_tree_schema_mode_enum_has_all_modes() { let schema = MemoryTreeTool.parameters_schema(); let modes: Vec<&str> = schema .get("properties") @@ -167,6 +194,7 @@ mod memory_tree_dispatcher_tests { assert!(modes.contains(&"query_global")); assert!(modes.contains(&"drill_down")); assert!(modes.contains(&"fetch_leaves")); + assert!(modes.contains(&"ingest_document")); } #[tokio::test]