mirror of
https://github.com/tinyhumansai/openhuman.git
synced 2026-07-27 21:08:00 +00:00
feat(memory): windowed cover_window tool + 24h morning brief (#3700)
Co-authored-by: Steven Enamakel <enamakel@tinyhumans.ai>
This commit is contained in:
sanil-23
co-authored by
Steven Enamakel
Steven Enamakel
parent
756b434821
commit
6b6db6129a
@@ -811,7 +811,9 @@ mod tests {
|
||||
let def = find("morning_briefing");
|
||||
assert_eq!(def.sandbox_mode, SandboxMode::ReadOnly);
|
||||
assert!(matches!(def.tools, ToolScope::Wildcard));
|
||||
assert!(!def.omit_memory_context);
|
||||
// The brief pulls its own last-24h memory via the `memory_tree`
|
||||
// `cover_window` tool, so the stale all-time memory blob is suppressed.
|
||||
assert!(def.omit_memory_context);
|
||||
assert!(def.omit_identity);
|
||||
assert!(def.omit_safety_preamble);
|
||||
assert_eq!(def.max_iterations, 8);
|
||||
|
||||
@@ -5,10 +5,14 @@ temperature = 0.5
|
||||
max_iterations = 8
|
||||
sandbox_mode = "read_only"
|
||||
|
||||
# Needs memory for user context and preferences, but not identity/safety
|
||||
# boilerplate — the prompt carries its own voice.
|
||||
# The brief pulls its recent memory itself via `memory_tree_cover_window`
|
||||
# (last 24h, source-grouped) — see prompt.md. We therefore suppress the
|
||||
# injected all-time memory blob (`## User Memory`, the namespace root
|
||||
# summaries): it is stale relative to "today" and would compete with the
|
||||
# fresh windowed tool result. Identity/safety boilerplate stays off too —
|
||||
# the prompt carries its own voice.
|
||||
omit_identity = true
|
||||
omit_memory_context = false
|
||||
omit_memory_context = true
|
||||
omit_safety_preamble = true
|
||||
omit_skills_catalog = true
|
||||
|
||||
|
||||
@@ -12,13 +12,13 @@ Prepare a morning briefing that helps the user start their day with clarity. Pul
|
||||
2. **Tasks & action items** — Open to-dos, deadlines due today, and anything overdue that needs attention.
|
||||
3. **Important emails / messages** — Unread threads that look time-sensitive or are from key contacts. Don't list every newsletter.
|
||||
4. **Crypto / market context** — If the user tracks markets, surface notable overnight moves, liquidation events, or governance votes closing today. Keep it to 2-3 bullets max.
|
||||
5. **Memory context** — Anything from recent memory that's relevant today (e.g. "you mentioned finishing the proposal by Wednesday" — and today is Wednesday).
|
||||
5. **Recent memory** — What actually happened across the user's connected sources in the **last 24 hours** (conversations, threads, activity), plus any commitment now due (e.g. "you said you'd finish the proposal by Wednesday" — and today is Wednesday).
|
||||
|
||||
## How to gather data
|
||||
|
||||
1. Use `composio_list_connections` to see what integrations the user has connected.
|
||||
2. For each relevant connection (calendar, email, task manager), use `composio_list_tools` to discover available actions, then `composio_execute` to pull today's data.
|
||||
3. Use memory context (already injected above) for user preferences, recurring patterns, and recent commitments.
|
||||
1. **Recent memory (last 24h).** Call the `memory_tree` tool with `mode: "cover_window"`, `since_ms = <now − 24h>` and `until_ms = <now>` (epoch-milliseconds — use the current date/time in the system context below to compute these). It returns the **minimum set of nodes** covering the window: condensed summaries where a whole stretch is in-window, and raw recent messages otherwise — grouped by source, oldest→newest. This is your authoritative recent-memory context; the all-time memory blob is intentionally NOT injected, so do not rely on it. Pass a `source_id`/`source_kind` filter if you only need one source.
|
||||
2. **Live data.** Use `composio_list_connections` to see connected integrations; for each relevant one (calendar, email, task manager), `composio_list_tools` then `composio_execute` to pull today's data.
|
||||
3. Reconcile the two: the 24h memory tells you what *happened*; the live calls tell you what's *scheduled / unread right now*. Don't double-report the same item.
|
||||
|
||||
## Tone & format
|
||||
|
||||
@@ -33,5 +33,5 @@ Prepare a morning briefing that helps the user start their day with clarity. Pul
|
||||
- **Never fabricate events, emails, or tasks.** Only include data you actually retrieved from tools or memory.
|
||||
- **Respect time zones.** The system prompt below carries the user's local date/time and IANA timezone — read it from there. Do **not** ask the user to repeat their timezone; only fall back to UTC and note it if the system context is genuinely missing the field.
|
||||
- **No stale data.** If a tool call fails or returns empty, say so — don't fall back to yesterday's data.
|
||||
- **Honor the timeline of memory.** Memory and prior-conversation context are stamped with when they were last updated (e.g. `(last updated 2026-05-25)`, `(as of …)`, `(noted …)`). Compare every such date against today's date in the system context. If an item predates the day you're briefing for, treat it as background — never restate an older daily summary, reminder, or "today you have…" note as if it is current. When the only relevant data is stale, name its date explicitly ("from your May 25 notes…") instead of presenting it as today's.
|
||||
- **Honor the timeline.** The `memory_tree` `cover_window` query already restricts recent memory to the last 24h, so treat its contents as genuinely recent. But each hit carries a real `time_range` — read it, and present things in the order they happened (oldest→newest). For anything carried over from a longer-lived note or a live tool result, compare its date against today's date in the system context: if it predates the day you're briefing for, name the date explicitly ("from your May 25 note…") rather than presenting it as today's.
|
||||
- **Privacy first.** Don't include full email bodies or message contents. Summarize senders and subjects.
|
||||
|
||||
@@ -0,0 +1,144 @@
|
||||
use crate::openhuman::config::rpc as config_rpc;
|
||||
use crate::openhuman::memory_store::chunks::types::SourceKind;
|
||||
use crate::openhuman::memory_tree::retrieval::cover::cover_window;
|
||||
use crate::openhuman::memory_tree::retrieval::rpc::CoverWindowRequest;
|
||||
use crate::openhuman::tools::traits::{Tool, ToolResult};
|
||||
use async_trait::async_trait;
|
||||
use serde_json::json;
|
||||
|
||||
/// Agent-facing wrapper for the windowed minimum-cover retrieval. Returns the
|
||||
/// smallest set of nodes (summaries + raw chunks) covering all memory in
|
||||
/// `[since_ms, until_ms]`. Built for time-bounded recaps like the morning
|
||||
/// brief's "last 24h" — see `memory_tree::retrieval::cover`.
|
||||
pub struct MemoryTreeCoverWindowTool;
|
||||
|
||||
#[async_trait]
|
||||
impl Tool for MemoryTreeCoverWindowTool {
|
||||
fn name(&self) -> &str {
|
||||
"memory_tree_cover_window"
|
||||
}
|
||||
|
||||
fn description(&self) -> &str {
|
||||
"Return the MINIMUM set of memory nodes covering a time window \
|
||||
[since_ms, until_ms] (epoch-milliseconds): condensed summaries where a \
|
||||
whole stretch is in-window, raw recent chunks otherwise. Grouped by \
|
||||
source, ordered oldest→newest. Use for time-bounded recaps (e.g. a \
|
||||
last-24h morning brief) instead of `query_source` (which is all-time)."
|
||||
}
|
||||
|
||||
fn parameters_schema(&self) -> serde_json::Value {
|
||||
json!({
|
||||
"type": "object",
|
||||
"properties": {
|
||||
"since_ms": {
|
||||
"type": "integer",
|
||||
"description": "Inclusive window start, epoch-milliseconds."
|
||||
},
|
||||
"until_ms": {
|
||||
"type": "integer",
|
||||
"description": "Inclusive window end, epoch-milliseconds."
|
||||
},
|
||||
"source_id": {
|
||||
"type": "string",
|
||||
"description": "Exact source id (e.g. `slack:#eng`, `gmail:abc`)."
|
||||
},
|
||||
"source_kind": {
|
||||
"type": "string",
|
||||
"enum": ["chat", "email", "document"],
|
||||
"description": "Source kind filter when no exact id is known."
|
||||
},
|
||||
"limit": {
|
||||
"type": "integer",
|
||||
"minimum": 0,
|
||||
"description": "Max hits to return (default 200)."
|
||||
}
|
||||
},
|
||||
"required": ["since_ms", "until_ms"]
|
||||
})
|
||||
}
|
||||
|
||||
async fn execute(&self, args: serde_json::Value) -> anyhow::Result<ToolResult> {
|
||||
log::debug!("[tool][memory_tree] cover_window invoked");
|
||||
let req: CoverWindowRequest = serde_json::from_value(args)
|
||||
.map_err(|e| anyhow::anyhow!("invalid arguments for memory_tree_cover_window: {e}"))?;
|
||||
// Correlation fields only — source_id can carry PII, so log its presence,
|
||||
// not its value.
|
||||
log::debug!(
|
||||
"[tool][memory_tree] cover_window parsed since_ms={} until_ms={} has_source_id={} has_source_kind={} has_limit={}",
|
||||
req.since_ms,
|
||||
req.until_ms,
|
||||
req.source_id.is_some(),
|
||||
req.source_kind.is_some(),
|
||||
req.limit.is_some()
|
||||
);
|
||||
let cfg = config_rpc::load_config_with_timeout()
|
||||
.await
|
||||
.map_err(|e| anyhow::anyhow!("memory_tree_cover_window: load config failed: {e}"))?;
|
||||
let source_kind = match req.source_kind.as_deref() {
|
||||
Some(s) => {
|
||||
log::trace!("[tool][memory_tree] cover_window parse_source_kind");
|
||||
Some(
|
||||
SourceKind::parse(s)
|
||||
.map_err(|e| anyhow::anyhow!("memory_tree_cover_window: {e}"))?,
|
||||
)
|
||||
}
|
||||
None => None,
|
||||
};
|
||||
log::trace!(
|
||||
"[tool][memory_tree] cover_window dispatch limit={}",
|
||||
req.limit.unwrap_or(0)
|
||||
);
|
||||
let resp = cover_window(
|
||||
&cfg,
|
||||
req.since_ms,
|
||||
req.until_ms,
|
||||
req.source_id.as_deref(),
|
||||
source_kind,
|
||||
req.limit.unwrap_or(0),
|
||||
)
|
||||
.await
|
||||
.map_err(|e| anyhow::anyhow!("memory_tree_cover_window: {e}"))?;
|
||||
log::debug!(
|
||||
"[tool][memory_tree] cover_window returning hits={} total={}",
|
||||
resp.hits.len(),
|
||||
resp.total
|
||||
);
|
||||
let json = serde_json::to_string(&resp)?;
|
||||
Ok(ToolResult::success(json))
|
||||
}
|
||||
}
|
||||
|
||||
#[cfg(test)]
|
||||
mod tests {
|
||||
use super::*;
|
||||
use crate::openhuman::tools::traits::Tool;
|
||||
use serde_json::json;
|
||||
|
||||
#[test]
|
||||
fn parameters_schema_requires_window_bounds() {
|
||||
let schema = MemoryTreeCoverWindowTool.parameters_schema();
|
||||
let required = schema.get("required").and_then(|r| r.as_array()).unwrap();
|
||||
assert!(required.iter().any(|v| v.as_str() == Some("since_ms")));
|
||||
assert!(required.iter().any(|v| v.as_str() == Some("until_ms")));
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn execute_rejects_missing_window_bounds() {
|
||||
let err = MemoryTreeCoverWindowTool
|
||||
.execute(json!({ "source_kind": "chat" }))
|
||||
.await
|
||||
.expect_err("missing since_ms/until_ms should fail");
|
||||
assert!(err
|
||||
.to_string()
|
||||
.contains("invalid arguments for memory_tree_cover_window"));
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn execute_rejects_invalid_source_kind() {
|
||||
let err = MemoryTreeCoverWindowTool
|
||||
.execute(json!({ "since_ms": 0, "until_ms": 1, "source_kind": "not-real" }))
|
||||
.await
|
||||
.expect_err("invalid source kind should fail");
|
||||
assert!(err.to_string().contains("memory_tree_cover_window:"));
|
||||
}
|
||||
}
|
||||
@@ -7,6 +7,7 @@
|
||||
//! the recommended single entry point for the `memory` orchestration layer.
|
||||
|
||||
mod backend;
|
||||
mod cover_window;
|
||||
mod drill_down;
|
||||
mod fetch_leaves;
|
||||
mod ingest_document;
|
||||
@@ -17,6 +18,7 @@ pub mod walk;
|
||||
|
||||
// Re-export individual tool types for callers that need them directly
|
||||
// (e.g. tool registration in ops.rs).
|
||||
pub use cover_window::MemoryTreeCoverWindowTool;
|
||||
pub use drill_down::MemoryTreeDrillDownTool;
|
||||
pub use fetch_leaves::MemoryTreeFetchLeavesTool;
|
||||
pub use ingest_document::MemoryTreeIngestDocumentTool;
|
||||
@@ -51,6 +53,7 @@ impl Tool for MemoryTreeTool {
|
||||
canonical id — call first when the user mentions someone by name), \
|
||||
`query_source` (filter by source type + time window), \
|
||||
`drill_down` (expand a coarse summary one level), \
|
||||
`cover_window` (minimum node set covering a time window [since_ms, until_ms] — use for last-24h / time-bounded recaps), \
|
||||
`fetch_leaves` (pull raw chunks for citation), `ingest_document` (write a document into the tree for future retrieval), \
|
||||
`walk` (agentic multi-turn walk — LLM navigates summaries and returns a synthesized answer for a natural-language query), \
|
||||
`smart_walk` (multi-strategy retrieval — combines vector search, keyword search, entity lookup, \
|
||||
@@ -64,10 +67,19 @@ impl Tool for MemoryTreeTool {
|
||||
"mode": {
|
||||
"type": "string",
|
||||
"enum": ["search_entities", "query_source",
|
||||
"drill_down", "fetch_leaves", "ingest_document", "walk",
|
||||
"drill_down", "cover_window", "fetch_leaves", "ingest_document", "walk",
|
||||
"smart_walk"],
|
||||
"description": "Which operation to run (retrieval or write)."
|
||||
},
|
||||
// cover_window params (epoch-milliseconds)
|
||||
"since_ms": {
|
||||
"type": "integer",
|
||||
"description": "cover_window: inclusive window start, epoch-milliseconds."
|
||||
},
|
||||
"until_ms": {
|
||||
"type": "integer",
|
||||
"description": "cover_window: inclusive window end, epoch-milliseconds."
|
||||
},
|
||||
// search_entities params
|
||||
"query": {
|
||||
"type": "string",
|
||||
@@ -143,6 +155,7 @@ impl Tool for MemoryTreeTool {
|
||||
"search_entities" => MemoryTreeSearchEntitiesTool.execute(args).await,
|
||||
"query_source" => MemoryTreeQuerySourceTool.execute(args).await,
|
||||
"drill_down" => MemoryTreeDrillDownTool.execute(args).await,
|
||||
"cover_window" => MemoryTreeCoverWindowTool.execute(args).await,
|
||||
"fetch_leaves" => MemoryTreeFetchLeavesTool.execute(args).await,
|
||||
"ingest_document" => MemoryTreeIngestDocumentTool.execute(args).await,
|
||||
"walk" => MemoryTreeWalkTool.execute(args).await,
|
||||
@@ -150,7 +163,7 @@ impl Tool for MemoryTreeTool {
|
||||
other => {
|
||||
log::debug!("[tool][memory_tree] unknown_mode mode={other}");
|
||||
Err(anyhow::anyhow!(
|
||||
"memory_tree: unknown mode `{other}`. Valid: search_entities, query_source, drill_down, fetch_leaves, ingest_document, walk, smart_walk"
|
||||
"memory_tree: unknown mode `{other}`. Valid: search_entities, query_source, drill_down, cover_window, fetch_leaves, ingest_document, walk, smart_walk"
|
||||
))
|
||||
}
|
||||
}
|
||||
@@ -193,6 +206,7 @@ mod memory_tree_dispatcher_tests {
|
||||
assert!(modes.contains(&"search_entities"));
|
||||
assert!(modes.contains(&"query_source"));
|
||||
assert!(modes.contains(&"drill_down"));
|
||||
assert!(modes.contains(&"cover_window"));
|
||||
assert!(modes.contains(&"fetch_leaves"));
|
||||
assert!(modes.contains(&"ingest_document"));
|
||||
assert!(modes.contains(&"walk"));
|
||||
|
||||
@@ -20,7 +20,9 @@ use crate::openhuman::memory_queue::types::{
|
||||
JobOutcome, NewJob, NodeRef, ReembedBackfillPayload, SealDocumentPayload, SealPayload,
|
||||
};
|
||||
use crate::openhuman::memory_store::chunks::store as chunk_store;
|
||||
use crate::openhuman::memory_store::chunks::types::{truncate_to_conservative_tokens, Chunk};
|
||||
use crate::openhuman::memory_store::chunks::types::{
|
||||
truncate_to_conservative_tokens, Chunk, Metadata,
|
||||
};
|
||||
use crate::openhuman::memory_store::content::{
|
||||
self as content_store, read as content_read, tags as content_tags,
|
||||
};
|
||||
@@ -69,6 +71,19 @@ fn derive_tree_scope(source_id: &str) -> String {
|
||||
source_id.to_string()
|
||||
}
|
||||
|
||||
/// The source-tree scope a chunk's content is appended under: its
|
||||
/// `path_scope` when set (shared-directory sources like Notion), otherwise the
|
||||
/// GitHub-aware [`derive_tree_scope`] of its `source_id`. This is the SAME
|
||||
/// mapping the append-buffer path uses (see `handle_extract`'s `AppendTarget`),
|
||||
/// so read paths such as `memory_tree::retrieval::cover` look up the tree the
|
||||
/// seal worker actually wrote to rather than the raw `source_id`.
|
||||
pub(crate) fn chunk_tree_scope(metadata: &Metadata) -> String {
|
||||
metadata
|
||||
.path_scope
|
||||
.clone()
|
||||
.unwrap_or_else(|| derive_tree_scope(&metadata.source_id))
|
||||
}
|
||||
|
||||
/// Whether a chunk's source uses the per-document rollup + versioning path
|
||||
/// (Notion). These chunks are deliberately **not** pushed into the flat L0
|
||||
/// buffer by `handle_extract` — their tree is built per document-version by
|
||||
@@ -307,11 +322,7 @@ fn finalize_extract(config: &Config, item: PreparedExtract) -> Result<JobOutcome
|
||||
chunk_id: chunk.id.clone(),
|
||||
},
|
||||
target: AppendTarget::Source {
|
||||
source_id: chunk
|
||||
.metadata
|
||||
.path_scope
|
||||
.clone()
|
||||
.unwrap_or_else(|| derive_tree_scope(&chunk.metadata.source_id)),
|
||||
source_id: chunk_tree_scope(&chunk.metadata),
|
||||
},
|
||||
})?)
|
||||
} else {
|
||||
|
||||
@@ -153,6 +153,7 @@ impl Tool for MemoryVectorSearchTool {
|
||||
until_ms: None,
|
||||
limit: Some(1000),
|
||||
source_scope: crate::openhuman::memory::source_scope::current_source_scope(),
|
||||
exclude_dropped: false,
|
||||
};
|
||||
|
||||
let chunks = list_chunks(&config, &query)
|
||||
|
||||
@@ -643,6 +643,11 @@ pub struct ListChunksQuery {
|
||||
/// prefix can't starve permitted rows. Non-source chunks always pass. `None`
|
||||
/// = unrestricted (the default for every non-agent caller).
|
||||
pub source_scope: Option<std::collections::HashSet<String>>,
|
||||
/// When `true`, rows the admission gate rejected (`lifecycle_status =
|
||||
/// 'dropped'`) are excluded. Default `false` preserves the all-rows
|
||||
/// behaviour every existing caller relies on; retrieval paths that must not
|
||||
/// surface filtered-out junk (e.g. `cover_window`) opt in.
|
||||
pub exclude_dropped: bool,
|
||||
}
|
||||
|
||||
/// List chunks matching the provided filters, ordered by `timestamp` DESC.
|
||||
@@ -676,6 +681,10 @@ pub fn list_chunks(config: &Config, query: &ListChunksQuery) -> Result<Vec<Chunk
|
||||
sql.push_str(" AND timestamp_ms <= ?");
|
||||
bound.push(Box::new(until_ms));
|
||||
}
|
||||
if query.exclude_dropped {
|
||||
sql.push_str(" AND lifecycle_status != ?");
|
||||
bound.push(Box::new(CHUNK_STATUS_DROPPED.to_string()));
|
||||
}
|
||||
let requested_limit = normalized_limit(query.limit);
|
||||
// When a profile source-scope is active, fetch a wider candidate set and
|
||||
// apply the gate in Rust *before* truncating, so a disallowed-source
|
||||
|
||||
@@ -129,6 +129,7 @@ impl RetrievalFacade {
|
||||
until_ms: filters.until_ms,
|
||||
limit: filters.limit,
|
||||
source_scope: None,
|
||||
exclude_dropped: false,
|
||||
};
|
||||
let rows = list_chunks(config, &query)?;
|
||||
let Some(required) = filters.tags_all_of.as_ref() else {
|
||||
|
||||
@@ -102,6 +102,7 @@ impl Tool for MemoryStoreRawChunksTool {
|
||||
until_ms: parsed.until_ms,
|
||||
limit: parsed.limit,
|
||||
source_scope: crate::openhuman::memory::source_scope::current_source_scope(),
|
||||
exclude_dropped: false,
|
||||
};
|
||||
let mut rows = list_chunks(&cfg, &query)?;
|
||||
if let Some(required) = parsed.tags_all_of.as_ref() {
|
||||
|
||||
@@ -795,6 +795,54 @@ pub fn list_summaries_at_level(
|
||||
})
|
||||
}
|
||||
|
||||
/// List every non-deleted summary in a tree whose time-range envelope is
|
||||
/// **fully contained** in `[since_ms, until_ms]` (inclusive), across all
|
||||
/// levels. This is the "eligible summary" set for the windowed minimum-cover
|
||||
/// (the morning-brief 24h tool): a node is eligible only when *all* its
|
||||
/// descendant leaves fall inside the window, which — because seal sets
|
||||
/// `time_range_start = MIN(children)` and `time_range_end = MAX(children)`
|
||||
/// (`bucket_seal::seal_one_level`) — is exactly
|
||||
/// `time_range_start_ms >= since_ms AND time_range_end_ms <= until_ms`.
|
||||
///
|
||||
/// Ordered by `level ASC, time_range_start_ms ASC` so callers can build the
|
||||
/// per-tree frontier and order chronologically without a re-sort. `content`
|
||||
/// here is the ≤500-char preview; callers that need the full body hydrate via
|
||||
/// `content::read::read_summary_body`.
|
||||
pub fn list_summaries_in_window(
|
||||
config: &Config,
|
||||
tree_id: &str,
|
||||
since_ms: i64,
|
||||
until_ms: i64,
|
||||
) -> Result<Vec<SummaryNode>> {
|
||||
log::debug!(
|
||||
"[tree::store] list_summaries_in_window tree_id={tree_id} since_ms={since_ms} until_ms={until_ms}"
|
||||
);
|
||||
with_connection(config, |conn| {
|
||||
let mut stmt = conn.prepare(
|
||||
"SELECT id, tree_id, tree_kind, level, parent_id,
|
||||
child_ids_json, content, token_count,
|
||||
entities_json, topics_json,
|
||||
time_range_start_ms, time_range_end_ms,
|
||||
score, sealed_at_ms, deleted, embedding,
|
||||
doc_id, version_ms
|
||||
FROM mem_tree_summaries
|
||||
WHERE tree_id = ?1 AND deleted = 0
|
||||
AND time_range_start_ms >= ?2
|
||||
AND time_range_end_ms <= ?3
|
||||
ORDER BY level ASC, time_range_start_ms ASC",
|
||||
)?;
|
||||
let rows = stmt
|
||||
.query_map(params![tree_id, since_ms, until_ms], row_to_summary)?
|
||||
.collect::<rusqlite::Result<Vec<_>>>()
|
||||
.context("Failed to collect in-window summaries")?;
|
||||
log::debug!(
|
||||
"[tree::store] list_summaries_in_window tree_id={tree_id} matched={}",
|
||||
rows.len()
|
||||
);
|
||||
Ok(rows)
|
||||
})
|
||||
}
|
||||
|
||||
/// Count summaries in a tree (diagnostic helper).
|
||||
pub fn count_summaries(config: &Config, tree_id: &str) -> Result<u64> {
|
||||
with_connection(config, |conn| {
|
||||
|
||||
@@ -88,6 +88,107 @@ fn summary_insert_and_fetch() {
|
||||
assert_eq!(count_summaries(&cfg, "tree-1").unwrap(), 1);
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn list_summaries_in_window_keeps_only_fully_contained() {
|
||||
// The cover's eligibility filter: a summary is returned only when its
|
||||
// ENTIRE envelope falls inside [since, until]. A node that straddles the
|
||||
// window edge (starts before `since`) must be excluded — using it would
|
||||
// drag in out-of-window content.
|
||||
let (_tmp, cfg) = test_config();
|
||||
insert_tree(&cfg, &sample_tree("tree-1", "slack:#eng")).unwrap();
|
||||
|
||||
let mk = |id: &str, start_ms: i64, end_ms: i64| {
|
||||
let mut n = sample_summary(id, "tree-1", 1);
|
||||
n.time_range_start = Utc.timestamp_millis_opt(start_ms).unwrap();
|
||||
n.time_range_end = Utc.timestamp_millis_opt(end_ms).unwrap();
|
||||
n
|
||||
};
|
||||
// window = [1000, 2000]
|
||||
let inside = mk("inside", 1100, 1900); // fully contained → eligible
|
||||
let straddle_start = mk("straddle", 900, 1500); // begins before since → excluded
|
||||
let straddle_end = mk("overrun", 1500, 2100); // ends after until → excluded
|
||||
let outside = mk("outside", 3000, 3500); // wholly after → excluded
|
||||
|
||||
with_connection(&cfg, |conn| {
|
||||
let tx = conn.unchecked_transaction()?;
|
||||
for n in [&inside, &straddle_start, &straddle_end, &outside] {
|
||||
insert_summary_tx(&tx, n, None, "test")?;
|
||||
}
|
||||
tx.commit()?;
|
||||
Ok(())
|
||||
})
|
||||
.unwrap();
|
||||
|
||||
let eligible = list_summaries_in_window(&cfg, "tree-1", 1000, 2000).unwrap();
|
||||
let ids: Vec<&str> = eligible.iter().map(|s| s.id.as_str()).collect();
|
||||
assert_eq!(
|
||||
ids,
|
||||
vec!["inside"],
|
||||
"only the fully-contained summary is eligible"
|
||||
);
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn list_summaries_in_window_includes_exact_boundaries() {
|
||||
// The window is inclusive on both ends: a summary whose envelope touches
|
||||
// `since`/`until` exactly is still fully contained, so it must be eligible.
|
||||
let (_tmp, cfg) = test_config();
|
||||
insert_tree(&cfg, &sample_tree("tree-1", "slack:#eng")).unwrap();
|
||||
|
||||
let mk = |id: &str, start_ms: i64, end_ms: i64| {
|
||||
let mut n = sample_summary(id, "tree-1", 1);
|
||||
n.time_range_start = Utc.timestamp_millis_opt(start_ms).unwrap();
|
||||
n.time_range_end = Utc.timestamp_millis_opt(end_ms).unwrap();
|
||||
n
|
||||
};
|
||||
// window = [1000, 2000]
|
||||
let start_on_edge = mk("start-edge", 1000, 1500); // starts exactly at `since`
|
||||
let end_on_edge = mk("end-edge", 1500, 2000); // ends exactly at `until`
|
||||
let both_edges = mk("both-edges", 1000, 2000); // spans the whole window
|
||||
|
||||
with_connection(&cfg, |conn| {
|
||||
let tx = conn.unchecked_transaction()?;
|
||||
for n in [&start_on_edge, &end_on_edge, &both_edges] {
|
||||
insert_summary_tx(&tx, n, None, "test")?;
|
||||
}
|
||||
tx.commit()?;
|
||||
Ok(())
|
||||
})
|
||||
.unwrap();
|
||||
|
||||
let eligible = list_summaries_in_window(&cfg, "tree-1", 1000, 2000).unwrap();
|
||||
let mut ids: Vec<&str> = eligible.iter().map(|s| s.id.as_str()).collect();
|
||||
ids.sort_unstable();
|
||||
assert_eq!(
|
||||
ids,
|
||||
vec!["both-edges", "end-edge", "start-edge"],
|
||||
"summaries touching the inclusive window edges are eligible"
|
||||
);
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn list_summaries_in_window_excludes_deleted() {
|
||||
let (_tmp, cfg) = test_config();
|
||||
insert_tree(&cfg, &sample_tree("tree-1", "slack:#eng")).unwrap();
|
||||
let mut node = sample_summary("sum-1", "tree-1", 1);
|
||||
node.time_range_start = Utc.timestamp_millis_opt(1100).unwrap();
|
||||
node.time_range_end = Utc.timestamp_millis_opt(1900).unwrap();
|
||||
node.deleted = true;
|
||||
with_connection(&cfg, |conn| {
|
||||
let tx = conn.unchecked_transaction()?;
|
||||
insert_summary_tx(&tx, &node, None, "test")?;
|
||||
tx.commit()?;
|
||||
Ok(())
|
||||
})
|
||||
.unwrap();
|
||||
assert!(
|
||||
list_summaries_in_window(&cfg, "tree-1", 1000, 2000)
|
||||
.unwrap()
|
||||
.is_empty(),
|
||||
"tombstoned summaries are never eligible"
|
||||
);
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn summary_insert_is_idempotent_on_id() {
|
||||
let (_tmp, cfg) = test_config();
|
||||
|
||||
@@ -0,0 +1,592 @@
|
||||
//! `memory_tree_cover_window` — the minimum-node cover of a time window.
|
||||
//!
|
||||
//! Given an explicit `[since_ms, until_ms]` window (and optional source
|
||||
//! filter), return the **smallest set of nodes that covers every in-window
|
||||
//! chunk** — a heterogeneous mix of summary nodes (where a whole subtree is
|
||||
//! in-window) and raw leaf chunks (everything else). This is the read path the
|
||||
//! morning brief uses for "last 24h" so it summarises only fresh content
|
||||
//! instead of the all-time root blob.
|
||||
//!
|
||||
//! The cover is **purely structural** — it does not assume the tree is
|
||||
//! calendar-bucketed (it isn't: `bucket_seal` groups by `SUMMARY_FANOUT`, so
|
||||
//! `level` is a depth integer, not hour/day). Two passes, per source tree:
|
||||
//!
|
||||
//! 1. **Eligible summaries** — every non-deleted summary whose time-range
|
||||
//! envelope is fully inside the window
|
||||
//! (`store::list_summaries_in_window`). Because seal sets the envelope to
|
||||
//! `MIN/MAX` of children, "envelope ⊆ window" ⇔ "all descendant leaves are
|
||||
//! in the window", i.e. using the summary drags in no out-of-window
|
||||
//! content.
|
||||
//! 2. **Frontier + raw fallback** — keep the topmost eligible summaries
|
||||
//! (`maximal` = eligible whose `parent_id` is not itself eligible); mark
|
||||
//! the chunks they cover (transitive descent through `child_ids`); emit any
|
||||
//! remaining in-window chunk raw. Raw chunks are the floor and cover
|
||||
//! boundary slices and not-yet-sealed content for free.
|
||||
//!
|
||||
//! Output is grouped by source and ordered ascending by start time so the
|
||||
//! brief reads it as a per-source timeline.
|
||||
|
||||
use std::collections::{HashMap, HashSet};
|
||||
|
||||
use anyhow::Result;
|
||||
|
||||
use crate::openhuman::config::Config;
|
||||
use crate::openhuman::memory_queue::handlers::chunk_tree_scope;
|
||||
use crate::openhuman::memory_store::chunks::store::{list_chunks, ListChunksQuery};
|
||||
use crate::openhuman::memory_store::chunks::types::{Chunk, SourceKind};
|
||||
use crate::openhuman::memory_store::content::read as content_read;
|
||||
use crate::openhuman::memory_store::trees::types::{SummaryNode, TreeKind};
|
||||
use crate::openhuman::memory_tree::retrieval::types::{
|
||||
hit_from_chunk, hit_from_summary, QueryResponse, RetrievalHit,
|
||||
};
|
||||
use crate::openhuman::memory_tree::tree::store;
|
||||
|
||||
/// Default cap on returned cover items when the caller passes `limit = 0`.
|
||||
const DEFAULT_LIMIT: usize = 200;
|
||||
|
||||
/// Upper bound on in-window chunks scanned across **all** sources. A 24h window
|
||||
/// rarely exceeds this; if it does we log and truncate (the excess simply
|
||||
/// doesn't appear — never silently mis-covered, since a frontier summary still
|
||||
/// stands in for a whole sealed subtree).
|
||||
const MAX_WINDOW_CHUNKS: usize = 5_000;
|
||||
|
||||
/// Per-source cap on raw in-window chunks retained for the cover. Applied
|
||||
/// **after** grouping so a single high-volume source can't crowd every other
|
||||
/// source out of the result; excess is logged, never silently mis-covered.
|
||||
const MAX_CHUNKS_PER_SOURCE: usize = 2_000;
|
||||
|
||||
/// Entrypoint for `memory_tree_cover_window`. Blocking SQLite work runs on
|
||||
/// `spawn_blocking`; the async caller stays on its runtime. Results are
|
||||
/// grouped by source (`tree_scope`) and ordered ascending by start time, then
|
||||
/// truncated to `limit` (`DEFAULT_LIMIT` when 0).
|
||||
pub async fn cover_window(
|
||||
config: &Config,
|
||||
since_ms: i64,
|
||||
until_ms: i64,
|
||||
source_id: Option<&str>,
|
||||
source_kind: Option<SourceKind>,
|
||||
limit: usize,
|
||||
) -> Result<QueryResponse> {
|
||||
let limit = if limit == 0 { DEFAULT_LIMIT } else { limit };
|
||||
if until_ms < since_ms {
|
||||
return Err(anyhow::anyhow!(
|
||||
"cover_window: until_ms ({until_ms}) precedes since_ms ({since_ms})"
|
||||
));
|
||||
}
|
||||
log::debug!(
|
||||
"[retrieval::cover] cover_window since_ms={since_ms} until_ms={until_ms} \
|
||||
has_source_id={} source_kind={:?} limit={limit}",
|
||||
source_id.is_some(),
|
||||
source_kind.map(|k| k.as_str()),
|
||||
);
|
||||
|
||||
let config_owned = config.clone();
|
||||
let source_id_owned = source_id.map(|s| s.to_string());
|
||||
// Capture the per-profile memory-source allowlist HERE, in the async task
|
||||
// that holds the `source_scope` task-local — `spawn_blocking` below runs on
|
||||
// a separate thread that does NOT inherit task-locals, so we must thread it
|
||||
// through explicitly or a restricted-profile brief would see every tagged
|
||||
// source. `None` = unrestricted.
|
||||
let source_scope = crate::openhuman::memory::source_scope::current_source_scope();
|
||||
let mut hits = tokio::task::spawn_blocking(move || -> Result<Vec<RetrievalHit>> {
|
||||
collect_cover(
|
||||
&config_owned,
|
||||
since_ms,
|
||||
until_ms,
|
||||
source_id_owned.as_deref(),
|
||||
source_kind,
|
||||
source_scope,
|
||||
)
|
||||
})
|
||||
.await
|
||||
.map_err(|e| anyhow::anyhow!("cover_window join error: {e}"))??;
|
||||
|
||||
// Group by source, then chronological ascending within each source. The
|
||||
// brief consumes this as a per-source timeline; ascending puts the
|
||||
// freshest content last (tool result lands at the context tail → recency).
|
||||
hits.sort_by(|a, b| {
|
||||
a.tree_scope
|
||||
.cmp(&b.tree_scope)
|
||||
.then(a.time_range_start.cmp(&b.time_range_start))
|
||||
});
|
||||
let total = hits.len();
|
||||
hits.truncate(limit);
|
||||
|
||||
log::debug!(
|
||||
"[retrieval::cover] returning hits={} total={}",
|
||||
hits.len(),
|
||||
total
|
||||
);
|
||||
Ok(QueryResponse::new(hits, total))
|
||||
}
|
||||
|
||||
/// Blocking: build the cover. **Chunk-driven, not tree-driven** — chunks are
|
||||
/// written to `mem_tree_chunks` at ingest, but the per-source `Tree` row is
|
||||
/// only created later by the seal worker. Iterating trees would therefore miss
|
||||
/// freshly-ingested (un-sealed) sources entirely — exactly the case the raw
|
||||
/// fallback exists for. So we pull the authoritative in-window chunk set first,
|
||||
/// group by source, and look up each source's tree (if any) for summaries.
|
||||
fn collect_cover(
|
||||
config: &Config,
|
||||
since_ms: i64,
|
||||
until_ms: i64,
|
||||
source_id: Option<&str>,
|
||||
source_kind: Option<SourceKind>,
|
||||
source_scope: Option<std::collections::HashSet<String>>,
|
||||
) -> Result<Vec<RetrievalHit>> {
|
||||
let chunks = list_chunks(
|
||||
config,
|
||||
&ListChunksQuery {
|
||||
source_id: source_id.map(|s| s.to_string()),
|
||||
source_kind,
|
||||
since_ms: Some(since_ms),
|
||||
until_ms: Some(until_ms),
|
||||
limit: Some(MAX_WINDOW_CHUNKS),
|
||||
// Skip rows the admission gate rejected: they linger in
|
||||
// `mem_tree_chunks` but were deliberately never appended to a tree,
|
||||
// so surfacing them raw would leak filtered-out junk into the brief.
|
||||
exclude_dropped: true,
|
||||
// Per-profile memory-source allowlist (threaded from the async task
|
||||
// — task-locals don't cross `spawn_blocking`).
|
||||
source_scope,
|
||||
..Default::default()
|
||||
},
|
||||
)?;
|
||||
if chunks.len() == MAX_WINDOW_CHUNKS {
|
||||
log::warn!(
|
||||
"[retrieval::cover] global in-window chunk cap {MAX_WINDOW_CHUNKS} hit — \
|
||||
some raw leaves may be omitted"
|
||||
);
|
||||
}
|
||||
|
||||
// Group by **tree scope**, not raw `source_id`: shared-directory sources
|
||||
// (Notion `path_scope`) and GitHub per-item ids seal under a derived scope,
|
||||
// so grouping by `source_id` would miss their tree and emit everything raw.
|
||||
// Use the same derivation as the append path. A per-source cap keeps one
|
||||
// high-volume source from crowding out the rest.
|
||||
let mut by_source: HashMap<String, Vec<Chunk>> = HashMap::new();
|
||||
let mut capped_sources = 0usize;
|
||||
let mut capped_chunks = 0usize;
|
||||
for chunk in chunks {
|
||||
let scope = chunk_tree_scope(&chunk.metadata);
|
||||
let bucket = by_source.entry(scope).or_default();
|
||||
if bucket.len() < MAX_CHUNKS_PER_SOURCE {
|
||||
bucket.push(chunk);
|
||||
} else {
|
||||
if bucket.len() == MAX_CHUNKS_PER_SOURCE {
|
||||
capped_sources += 1;
|
||||
}
|
||||
capped_chunks += 1;
|
||||
}
|
||||
}
|
||||
if capped_chunks > 0 {
|
||||
// Omit scope (PII) — aggregate counts only.
|
||||
log::warn!(
|
||||
"[retrieval::cover] {capped_sources} source(s) hit per-source cap \
|
||||
{MAX_CHUNKS_PER_SOURCE} — {capped_chunks} raw leaf/leaves omitted"
|
||||
);
|
||||
}
|
||||
log::debug!("[retrieval::cover] in-window sources n={}", by_source.len());
|
||||
|
||||
// An exact `source_id` filter means `chunks` is a strict subset of its
|
||||
// (possibly shared) tree, so shared-tree summaries must be restricted to
|
||||
// the requested leaves. Without a filter every in-window leaf is present.
|
||||
let exact_source = source_id.is_some();
|
||||
let mut hits: Vec<RetrievalHit> = Vec::new();
|
||||
for (source, src_chunks) in by_source {
|
||||
cover_one_source(
|
||||
config,
|
||||
&source,
|
||||
since_ms,
|
||||
until_ms,
|
||||
src_chunks,
|
||||
exact_source,
|
||||
&mut hits,
|
||||
)?;
|
||||
}
|
||||
Ok(hits)
|
||||
}
|
||||
|
||||
/// Minimum cover for one source: frontier summaries (when the source has a
|
||||
/// sealed tree) plus every in-window chunk they don't already cover, raw. A
|
||||
/// source with no `Tree` row (not yet processed by the seal worker) has no
|
||||
/// eligible summaries, so all its in-window chunks are emitted raw.
|
||||
fn cover_one_source(
|
||||
config: &Config,
|
||||
source: &str,
|
||||
since_ms: i64,
|
||||
until_ms: i64,
|
||||
chunks: Vec<Chunk>,
|
||||
exact_source: bool,
|
||||
out: &mut Vec<RetrievalHit>,
|
||||
) -> Result<()> {
|
||||
// Look up the source's summary tree (scope == source id). Absent until the
|
||||
// seal worker first processes this source.
|
||||
let tree = store::get_tree_by_scope(config, TreeKind::Source, source)?;
|
||||
let (tree_id, eligible) = match &tree {
|
||||
Some(t) => (
|
||||
t.id.as_str(),
|
||||
store::list_summaries_in_window(config, &t.id, since_ms, until_ms)?,
|
||||
),
|
||||
None => ("", Vec::new()),
|
||||
};
|
||||
// Latest-wins for versioned document sources (Notion): drop superseded
|
||||
// doc-root revisions so the cover never emits a stale page, and remember
|
||||
// their chunk ids so the raw fallback below doesn't resurface them either.
|
||||
let (eligible, suppressed_chunk_ids) = filter_superseded_doc_versions(eligible);
|
||||
// In exact-source mode the tree may be shared across sibling sources, so
|
||||
// only emit summaries whose whole subtree is among the filtered chunks.
|
||||
let present: HashSet<&str> = chunks.iter().map(|c| c.id.as_str()).collect();
|
||||
let plan = plan_cover(&eligible, exact_source.then_some(&present));
|
||||
|
||||
// Frontier summaries (hydrated to full body).
|
||||
let by_id: HashMap<&str, &SummaryNode> = eligible.iter().map(|s| (s.id.as_str(), s)).collect();
|
||||
for id in &plan.maximal_ids {
|
||||
let Some(node) = by_id.get(id.as_str()) else {
|
||||
continue;
|
||||
};
|
||||
let mut node = (*node).clone();
|
||||
match content_read::read_summary_body(config, &node.id) {
|
||||
Ok(body) => node.content = body,
|
||||
Err(e) => {
|
||||
log::warn!("[retrieval::cover] read_summary_body failed — serving preview: {e:#}")
|
||||
}
|
||||
}
|
||||
out.push(hit_from_summary(&node, source));
|
||||
}
|
||||
|
||||
// In-window chunks not covered by a frontier summary, raw — skipping any
|
||||
// chunk under a superseded document revision.
|
||||
for chunk in &chunks {
|
||||
if plan.covered_chunk_ids.contains(&chunk.id) || suppressed_chunk_ids.contains(&chunk.id) {
|
||||
continue;
|
||||
}
|
||||
let mut chunk = chunk.clone();
|
||||
match content_read::read_chunk_body(config, &chunk.id) {
|
||||
Ok(body) => chunk.content = body,
|
||||
Err(e) => {
|
||||
log::warn!("[retrieval::cover] read_chunk_body failed — serving preview: {e:#}")
|
||||
}
|
||||
}
|
||||
out.push(hit_from_chunk(&chunk, tree_id, source, 0.0));
|
||||
}
|
||||
Ok(())
|
||||
}
|
||||
|
||||
/// The structural result of the cover for one tree's eligible summaries:
|
||||
/// which summaries to emit (the frontier) and which chunk ids they already
|
||||
/// cover (so the caller can emit the rest raw). Pure — no I/O — so the cover
|
||||
/// logic is unit-testable without a database.
|
||||
struct CoverPlan {
|
||||
/// Topmost eligible summary ids (eligible nodes whose parent is not
|
||||
/// eligible). These stand in for their whole subtree.
|
||||
maximal_ids: Vec<String>,
|
||||
/// Leaf chunk ids transitively covered by the `maximal` summaries.
|
||||
covered_chunk_ids: HashSet<String>,
|
||||
}
|
||||
|
||||
/// Compute the frontier + covered-chunk set from a tree's eligible summaries.
|
||||
///
|
||||
/// `eligible` must be exactly the summaries whose envelope is inside the
|
||||
/// window (all levels). A summary is **maximal** when its `parent_id` is not
|
||||
/// itself eligible. Coverage descends `child_ids`: an id that is not another
|
||||
/// eligible summary is a leaf chunk id (because every descendant of a
|
||||
/// fully-covered node is itself fully covered, hence eligible — so any child
|
||||
/// not in the eligible set is a leaf).
|
||||
///
|
||||
/// `restrict_to_present` guards the exact-source path. A source tree's scope
|
||||
/// can be **broader** than the requested `source_id` (Notion `path_scope`,
|
||||
/// GitHub repo-scoped trees seal many pages/issues into one tree). When the
|
||||
/// caller filtered chunks down to a single source id, a frontier summary over
|
||||
/// the shared tree would also cover *sibling* sources' leaves — leaking
|
||||
/// unrelated memory and masking the requested raw chunks under a mixed
|
||||
/// summary. When `Some(present)`, a maximal summary is emitted only if **every**
|
||||
/// chunk it covers is in `present` (the in-filter chunk ids); summaries that
|
||||
/// span out-of-filter sources are dropped, and their in-filter chunks fall
|
||||
/// through to raw emission. `None` (the no-filter brief path) keeps every
|
||||
/// maximal summary — the chunk set already holds every in-window leaf, so no
|
||||
/// summary can span anything absent.
|
||||
fn plan_cover(eligible: &[SummaryNode], restrict_to_present: Option<&HashSet<&str>>) -> CoverPlan {
|
||||
let eligible_ids: HashSet<&str> = eligible.iter().map(|s| s.id.as_str()).collect();
|
||||
let by_id: HashMap<&str, &SummaryNode> = eligible.iter().map(|s| (s.id.as_str(), s)).collect();
|
||||
|
||||
let mut maximal_ids: Vec<String> = Vec::new();
|
||||
let mut covered_chunk_ids: HashSet<String> = HashSet::new();
|
||||
for node in eligible.iter().filter(|s| match &s.parent_id {
|
||||
Some(parent) => !eligible_ids.contains(parent.as_str()),
|
||||
None => true,
|
||||
}) {
|
||||
let mut sub: HashSet<String> = HashSet::new();
|
||||
collect_descendant_chunks(node, &by_id, &mut sub);
|
||||
if let Some(present) = restrict_to_present {
|
||||
if !sub.iter().all(|c| present.contains(c.as_str())) {
|
||||
// Shared-tree summary spans sources outside the exact-source
|
||||
// filter — skip it; its in-filter chunks are emitted raw.
|
||||
continue;
|
||||
}
|
||||
}
|
||||
maximal_ids.push(node.id.clone());
|
||||
covered_chunk_ids.extend(sub);
|
||||
}
|
||||
|
||||
CoverPlan {
|
||||
maximal_ids,
|
||||
covered_chunk_ids,
|
||||
}
|
||||
}
|
||||
|
||||
/// Walk a summary's subtree (within the eligible set) collecting leaf chunk
|
||||
/// ids. A child id present in `by_id` is a lower-level summary → recurse; a
|
||||
/// child id absent from `by_id` is a leaf chunk → record it.
|
||||
fn collect_descendant_chunks(
|
||||
node: &SummaryNode,
|
||||
by_id: &HashMap<&str, &SummaryNode>,
|
||||
covered: &mut HashSet<String>,
|
||||
) {
|
||||
for child in &node.child_ids {
|
||||
match by_id.get(child.as_str()) {
|
||||
Some(child_summary) => collect_descendant_chunks(child_summary, by_id, covered),
|
||||
None => {
|
||||
covered.insert(child.clone());
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
/// Latest-wins for versioned document sources (Notion). A source tree can hold
|
||||
/// several doc-root summaries for the **same** `doc_id` — each page edit seals
|
||||
/// a new doc-root at a higher `version_ms` beside the old one, and retrieval is
|
||||
/// expected to hide the older revisions (`drill_down` does the same). Returns
|
||||
/// `eligible` with every superseded revision's whole subtree removed, plus the
|
||||
/// chunk ids under those dropped revisions so the raw fallback can't resurface
|
||||
/// stale page content. Summaries with no `doc_id` are untouched.
|
||||
fn filter_superseded_doc_versions(
|
||||
eligible: Vec<SummaryNode>,
|
||||
) -> (Vec<SummaryNode>, HashSet<String>) {
|
||||
// No document nodes → nothing to do (the common chat/email case).
|
||||
if !eligible.iter().any(|s| s.doc_id.is_some()) {
|
||||
return (eligible, HashSet::new());
|
||||
}
|
||||
|
||||
let by_id: HashMap<&str, &SummaryNode> = eligible.iter().map(|s| (s.id.as_str(), s)).collect();
|
||||
|
||||
// Winning (max) version per doc_id; `version_ms` defaults to i64::MIN so a
|
||||
// legacy untagged doc-root never wins over a tagged one.
|
||||
let mut max_version_by_doc: HashMap<&str, i64> = HashMap::new();
|
||||
for s in &eligible {
|
||||
if let Some(doc) = s.doc_id.as_deref() {
|
||||
let v = s.version_ms.unwrap_or(i64::MIN);
|
||||
max_version_by_doc
|
||||
.entry(doc)
|
||||
.and_modify(|m| {
|
||||
if v > *m {
|
||||
*m = v;
|
||||
}
|
||||
})
|
||||
.or_insert(v);
|
||||
}
|
||||
}
|
||||
|
||||
// A doc-root is a "loser" when it's an older revision, or a duplicate of the
|
||||
// winning version (e.g. a retried SealDocument minted two) — keep the first
|
||||
// winner only. `eligible` is ordered (level, start), so dedup is stable.
|
||||
let mut winners_seen: HashSet<&str> = HashSet::new();
|
||||
let mut removed_summary_ids: HashSet<String> = HashSet::new();
|
||||
let mut suppressed_chunk_ids: HashSet<String> = HashSet::new();
|
||||
for s in &eligible {
|
||||
let Some(doc) = s.doc_id.as_deref() else {
|
||||
continue;
|
||||
};
|
||||
let v = s.version_ms.unwrap_or(i64::MIN);
|
||||
let max = max_version_by_doc.get(doc).copied().unwrap_or(i64::MIN);
|
||||
// Short-circuit keeps the winner slot untouched for older revisions.
|
||||
let loser = v < max || !winners_seen.insert(doc);
|
||||
if loser {
|
||||
removed_summary_ids.insert(s.id.clone());
|
||||
collect_subtree_ids(
|
||||
s,
|
||||
&by_id,
|
||||
&mut removed_summary_ids,
|
||||
&mut suppressed_chunk_ids,
|
||||
);
|
||||
}
|
||||
}
|
||||
|
||||
let kept = eligible
|
||||
.into_iter()
|
||||
.filter(|s| !removed_summary_ids.contains(&s.id))
|
||||
.collect();
|
||||
(kept, suppressed_chunk_ids)
|
||||
}
|
||||
|
||||
/// Walk a summary's subtree collecting both descendant summary ids (into
|
||||
/// `summaries`) and leaf chunk ids (into `chunks`). Used to evict a superseded
|
||||
/// document revision's whole subtree from the cover.
|
||||
fn collect_subtree_ids(
|
||||
node: &SummaryNode,
|
||||
by_id: &HashMap<&str, &SummaryNode>,
|
||||
summaries: &mut HashSet<String>,
|
||||
chunks: &mut HashSet<String>,
|
||||
) {
|
||||
for child in &node.child_ids {
|
||||
match by_id.get(child.as_str()) {
|
||||
Some(child_summary) => {
|
||||
summaries.insert(child.clone());
|
||||
collect_subtree_ids(child_summary, by_id, summaries, chunks);
|
||||
}
|
||||
None => {
|
||||
chunks.insert(child.clone());
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
#[cfg(test)]
|
||||
mod tests {
|
||||
use super::*;
|
||||
use crate::openhuman::memory_store::trees::types::TreeKind;
|
||||
use chrono::Utc;
|
||||
|
||||
fn summary(id: &str, parent: Option<&str>, level: u32, children: &[&str]) -> SummaryNode {
|
||||
SummaryNode {
|
||||
id: id.to_string(),
|
||||
tree_id: "t1".to_string(),
|
||||
tree_kind: TreeKind::Source,
|
||||
level,
|
||||
parent_id: parent.map(|p| p.to_string()),
|
||||
child_ids: children.iter().map(|c| c.to_string()).collect(),
|
||||
content: format!("summary {id}"),
|
||||
token_count: 10,
|
||||
entities: vec![],
|
||||
topics: vec![],
|
||||
time_range_start: Utc::now(),
|
||||
time_range_end: Utc::now(),
|
||||
score: 0.0,
|
||||
sealed_at: Utc::now(),
|
||||
deleted: false,
|
||||
embedding: None,
|
||||
doc_id: None,
|
||||
version_ms: None,
|
||||
}
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn single_eligible_summary_covers_its_leaves() {
|
||||
// L1 node over two leaf chunks, no parent → it's the frontier and
|
||||
// covers both leaves.
|
||||
let eligible = vec![summary("s1", None, 1, &["chunk-a", "chunk-b"])];
|
||||
let plan = plan_cover(&eligible, None);
|
||||
assert_eq!(plan.maximal_ids, vec!["s1"]);
|
||||
assert!(plan.covered_chunk_ids.contains("chunk-a"));
|
||||
assert!(plan.covered_chunk_ids.contains("chunk-b"));
|
||||
assert_eq!(plan.covered_chunk_ids.len(), 2);
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn parent_subsumes_child_only_parent_is_maximal() {
|
||||
// s2 (L2) → s1 (L1) → leaves. Both eligible: only s2 is maximal, and
|
||||
// it transitively covers the leaves under s1.
|
||||
let eligible = vec![
|
||||
summary("s2", None, 2, &["s1"]),
|
||||
summary("s1", Some("s2"), 1, &["chunk-a", "chunk-b"]),
|
||||
];
|
||||
let plan = plan_cover(&eligible, None);
|
||||
assert_eq!(plan.maximal_ids, vec!["s2"]);
|
||||
assert!(plan.covered_chunk_ids.contains("chunk-a"));
|
||||
assert!(plan.covered_chunk_ids.contains("chunk-b"));
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn ineligible_parent_leaves_child_as_frontier() {
|
||||
// s1 is eligible but its parent s2 is NOT in the eligible set (it
|
||||
// straddles the window / not sealed) → s1 is maximal. Leaves under s1
|
||||
// are covered; nothing else.
|
||||
let eligible = vec![summary("s1", Some("s2-not-eligible"), 1, &["chunk-a"])];
|
||||
let plan = plan_cover(&eligible, None);
|
||||
assert_eq!(plan.maximal_ids, vec!["s1"]);
|
||||
assert!(plan.covered_chunk_ids.contains("chunk-a"));
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn empty_eligible_set_covers_nothing() {
|
||||
// No sealed/eligible summaries → frontier empty, no chunk covered, so
|
||||
// the caller emits every in-window chunk raw.
|
||||
let plan = plan_cover(&[], None);
|
||||
assert!(plan.maximal_ids.is_empty());
|
||||
assert!(plan.covered_chunk_ids.is_empty());
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn sibling_frontier_nodes_each_emitted() {
|
||||
// Two L1 siblings, both eligible, parent NOT eligible → both maximal.
|
||||
let eligible = vec![
|
||||
summary("s1", Some("root-x"), 1, &["chunk-a"]),
|
||||
summary("s2", Some("root-x"), 1, &["chunk-b"]),
|
||||
];
|
||||
let mut plan = plan_cover(&eligible, None);
|
||||
plan.maximal_ids.sort();
|
||||
assert_eq!(plan.maximal_ids, vec!["s1", "s2"]);
|
||||
assert_eq!(plan.covered_chunk_ids.len(), 2);
|
||||
}
|
||||
|
||||
fn doc_summary(id: &str, doc_id: &str, version_ms: i64, children: &[&str]) -> SummaryNode {
|
||||
let mut s = summary(id, Some("merge-root"), 1, children);
|
||||
s.doc_id = Some(doc_id.to_string());
|
||||
s.version_ms = Some(version_ms);
|
||||
s
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn filter_superseded_doc_versions_keeps_newest_and_suppresses_old_chunks() {
|
||||
// Two revisions of the same Notion page plus an unrelated chat summary.
|
||||
// Only the newest revision survives; the older revision's subtree is
|
||||
// dropped and its chunk reported for raw-fallback suppression.
|
||||
let eligible = vec![
|
||||
doc_summary("pageA@v1", "notion:pageA", 100, &["chunk-old"]),
|
||||
doc_summary("pageA@v2", "notion:pageA", 200, &["chunk-new"]),
|
||||
summary("chat", Some("root"), 1, &["chunk-chat"]),
|
||||
];
|
||||
let (kept, suppressed) = filter_superseded_doc_versions(eligible);
|
||||
let kept_ids: Vec<&str> = kept.iter().map(|s| s.id.as_str()).collect();
|
||||
assert_eq!(kept_ids, vec!["pageA@v2", "chat"]);
|
||||
assert!(suppressed.contains("chunk-old"));
|
||||
assert!(!suppressed.contains("chunk-new"));
|
||||
assert!(!suppressed.contains("chunk-chat"));
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn filter_superseded_doc_versions_dedups_duplicate_winning_revision() {
|
||||
// A retried seal can mint two doc-roots at the SAME winning version;
|
||||
// keep the first, drop the duplicate's subtree.
|
||||
let eligible = vec![
|
||||
doc_summary("dup-a", "notion:pageB", 300, &["chunk-a"]),
|
||||
doc_summary("dup-b", "notion:pageB", 300, &["chunk-b"]),
|
||||
];
|
||||
let (kept, suppressed) = filter_superseded_doc_versions(eligible);
|
||||
let kept_ids: Vec<&str> = kept.iter().map(|s| s.id.as_str()).collect();
|
||||
assert_eq!(kept_ids, vec!["dup-a"]);
|
||||
assert!(suppressed.contains("chunk-b"));
|
||||
assert!(!suppressed.contains("chunk-a"));
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn restrict_drops_summaries_spanning_out_of_filter_chunks() {
|
||||
// Exact-source mode over a SHARED tree: `s_mixed` summarises a leaf the
|
||||
// filter kept (`chunk-a`) plus a sibling page's leaf (`chunk-foreign`)
|
||||
// that isn't in the requested set; `s_clean` covers only kept leaves.
|
||||
// With the present-set restriction, the mixed summary must be dropped
|
||||
// (so `chunk-a` falls through to raw) while the clean one survives.
|
||||
let eligible = vec![
|
||||
summary("s_mixed", Some("root"), 1, &["chunk-a", "chunk-foreign"]),
|
||||
summary("s_clean", Some("root"), 1, &["chunk-b"]),
|
||||
];
|
||||
let present: HashSet<&str> = ["chunk-a", "chunk-b"].into_iter().collect();
|
||||
let plan = plan_cover(&eligible, Some(&present));
|
||||
assert_eq!(plan.maximal_ids, vec!["s_clean"]);
|
||||
assert!(plan.covered_chunk_ids.contains("chunk-b"));
|
||||
// chunk-a is NOT covered → caller emits it raw rather than via the
|
||||
// sibling-spanning summary.
|
||||
assert!(!plan.covered_chunk_ids.contains("chunk-a"));
|
||||
assert!(!plan.covered_chunk_ids.contains("chunk-foreign"));
|
||||
}
|
||||
}
|
||||
@@ -12,10 +12,12 @@
|
||||
//! - `openhuman.memory_tree_search_entities` — fuzzy canonical-id lookup
|
||||
//! - `openhuman.memory_tree_drill_down` — walk summary children
|
||||
//! - `openhuman.memory_tree_fetch_leaves` — batch chunk hydration
|
||||
//! - `openhuman.memory_tree_cover_window` — minimum-node cover of a window
|
||||
//!
|
||||
//! All tools share the [`types::RetrievalHit`] / [`types::QueryResponse`]
|
||||
//! shape so the LLM sees a uniform schema regardless of which tool ran.
|
||||
|
||||
pub mod cover;
|
||||
pub mod drill_down;
|
||||
pub mod fetch;
|
||||
pub mod rpc;
|
||||
@@ -29,6 +31,7 @@ mod benchmarks;
|
||||
#[cfg(test)]
|
||||
mod integration_tests;
|
||||
|
||||
pub use cover::cover_window;
|
||||
pub use drill_down::drill_down;
|
||||
pub use fetch::fetch_leaves;
|
||||
pub use schemas::{
|
||||
|
||||
@@ -10,6 +10,7 @@ use serde::{Deserialize, Serialize};
|
||||
use crate::openhuman::config::Config;
|
||||
use crate::openhuman::memory_store::chunks::types::SourceKind;
|
||||
use crate::openhuman::memory_tree::retrieval::{
|
||||
cover::cover_window,
|
||||
drill_down::drill_down,
|
||||
fetch::fetch_leaves,
|
||||
search::search_entities,
|
||||
@@ -77,6 +78,76 @@ pub async fn query_source_rpc(
|
||||
))
|
||||
}
|
||||
|
||||
// ── cover_window ──────────────────────────────────────────────────────
|
||||
|
||||
/// Request body for `memory_tree_cover_window`. `since_ms`/`until_ms` are the
|
||||
/// inclusive window bounds in epoch-milliseconds; the source filter mirrors
|
||||
/// `query_source`. See [`super::cover::cover_window`] for cover semantics.
|
||||
#[derive(Clone, Debug, Default, Serialize, Deserialize)]
|
||||
pub struct CoverWindowRequest {
|
||||
pub since_ms: i64,
|
||||
pub until_ms: i64,
|
||||
#[serde(default)]
|
||||
pub source_id: Option<String>,
|
||||
#[serde(default)]
|
||||
pub source_kind: Option<String>,
|
||||
#[serde(default)]
|
||||
pub limit: Option<usize>,
|
||||
}
|
||||
|
||||
/// JSON-RPC handler body for `memory_tree_cover_window`. Parses the request,
|
||||
/// delegates to [`super::cover::cover_window`], logs PII-redacted counts.
|
||||
pub async fn cover_window_rpc(
|
||||
config: &Config,
|
||||
req: CoverWindowRequest,
|
||||
) -> Result<RpcOutcome<QueryResponse>, String> {
|
||||
log::debug!(
|
||||
"[rpc][memory_tree] cover_window enter since_ms={} until_ms={} has_source_id={} has_source_kind={} has_limit={}",
|
||||
req.since_ms,
|
||||
req.until_ms,
|
||||
req.source_id.is_some(),
|
||||
req.source_kind.is_some(),
|
||||
req.limit.is_some()
|
||||
);
|
||||
let source_kind = match req.source_kind.as_deref() {
|
||||
Some(s) => {
|
||||
log::trace!("[rpc][memory_tree] cover_window parse_source_kind");
|
||||
Some(SourceKind::parse(s).map_err(|e| format!("cover_window: {e}"))?)
|
||||
}
|
||||
None => None,
|
||||
};
|
||||
let limit = req.limit.unwrap_or(0);
|
||||
log::trace!("[rpc][memory_tree] cover_window dispatch limit={limit}");
|
||||
let resp = cover_window(
|
||||
config,
|
||||
req.since_ms,
|
||||
req.until_ms,
|
||||
req.source_id.as_deref(),
|
||||
source_kind,
|
||||
limit,
|
||||
)
|
||||
.await
|
||||
.map_err(|e| format!("cover_window: {e}"))?;
|
||||
let n = resp.hits.len();
|
||||
log::debug!(
|
||||
"[rpc][memory_tree] cover_window exit hits={} total={}",
|
||||
n,
|
||||
resp.total
|
||||
);
|
||||
// Omit scope / source_id from the log — can carry PII. Counts only.
|
||||
Ok(RpcOutcome::single_log(
|
||||
resp,
|
||||
format!(
|
||||
"memory_tree: cover_window since_ms={} until_ms={} has_source_id={} source_kind={:?} hits={}",
|
||||
req.since_ms,
|
||||
req.until_ms,
|
||||
req.source_id.is_some(),
|
||||
req.source_kind,
|
||||
n
|
||||
),
|
||||
))
|
||||
}
|
||||
|
||||
// ── search_entities ───────────────────────────────────────────────────
|
||||
|
||||
/// Request body for `memory_tree_search_entities`.
|
||||
@@ -331,6 +402,104 @@ mod tests {
|
||||
assert!(err.contains("unknown source kind: bogus"), "got {err}");
|
||||
}
|
||||
|
||||
// ── cover_window_rpc ──────────────────────────────────────────────
|
||||
|
||||
#[tokio::test]
|
||||
async fn cover_window_rpc_returns_empty_with_no_data_and_redacts_log() {
|
||||
let (_tmp, cfg) = test_config();
|
||||
let req = CoverWindowRequest {
|
||||
since_ms: 0,
|
||||
until_ms: 4_000_000_000_000,
|
||||
source_id: Some("slack:#eng".into()),
|
||||
source_kind: Some("chat".into()),
|
||||
limit: None,
|
||||
};
|
||||
let outcome = cover_window_rpc(&cfg, req).await.unwrap();
|
||||
assert!(outcome.value.hits.is_empty());
|
||||
assert_eq!(outcome.value.total, 0);
|
||||
assert_eq!(outcome.logs.len(), 1);
|
||||
let log = &outcome.logs[0];
|
||||
assert!(log.contains("has_source_id=true"), "log: {log}");
|
||||
assert!(log.contains("source_kind=Some(\"chat\")"), "log: {log}");
|
||||
assert!(log.contains("hits=0"), "log: {log}");
|
||||
// PII redaction: the raw source_id must NOT leak into the log.
|
||||
assert!(!log.contains("slack:#eng"), "log leaked source_id: {log}");
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn cover_window_rpc_rejects_invalid_source_kind() {
|
||||
let (_tmp, cfg) = test_config();
|
||||
let req = CoverWindowRequest {
|
||||
since_ms: 0,
|
||||
until_ms: 1,
|
||||
source_id: None,
|
||||
source_kind: Some("bogus".into()),
|
||||
limit: None,
|
||||
};
|
||||
let err = cover_window_rpc(&cfg, req).await.unwrap_err();
|
||||
assert!(err.contains("cover_window:"), "got {err}");
|
||||
assert!(err.contains("unknown source kind: bogus"), "got {err}");
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn cover_window_rpc_honors_profile_source_scope() {
|
||||
use crate::openhuman::memory::source_scope::with_source_scope;
|
||||
let (_tmp, cfg) = test_config();
|
||||
// Two memory-source chunks in different sources, both inside the window.
|
||||
let mut allowed = sample_chunk("slack:#eng", 0);
|
||||
allowed.metadata.tags = vec!["memory_sources".into(), "chat".into()];
|
||||
let mut blocked = sample_chunk("slack:#secret", 0);
|
||||
blocked.metadata.tags = vec!["memory_sources".into(), "chat".into()];
|
||||
upsert_chunks(&cfg, &[allowed.clone(), blocked.clone()]).unwrap();
|
||||
stage_test_chunks(&cfg, &[allowed.clone(), blocked.clone()]);
|
||||
|
||||
let req = || CoverWindowRequest {
|
||||
since_ms: 0,
|
||||
until_ms: 4_000_000_000_000,
|
||||
source_id: None,
|
||||
source_kind: None,
|
||||
limit: None,
|
||||
};
|
||||
|
||||
// A restricted profile (allowlist = #eng only) must not surface #secret,
|
||||
// even though cover_window does its DB work on a spawn_blocking thread
|
||||
// that does not inherit the source_scope task-local.
|
||||
let resp = with_source_scope(Some(vec!["slack:#eng".into()]), async {
|
||||
cover_window_rpc(&cfg, req()).await
|
||||
})
|
||||
.await
|
||||
.unwrap();
|
||||
let ids: Vec<&str> = resp.value.hits.iter().map(|h| h.node_id.as_str()).collect();
|
||||
assert!(
|
||||
ids.contains(&allowed.id.as_str()),
|
||||
"allowlisted source must be present: {ids:?}"
|
||||
);
|
||||
assert!(
|
||||
!ids.contains(&blocked.id.as_str()),
|
||||
"disallowed source must be filtered out: {ids:?}"
|
||||
);
|
||||
|
||||
// With no profile scope active, both sources are visible.
|
||||
let unrestricted = cover_window_rpc(&cfg, req()).await.unwrap();
|
||||
assert_eq!(unrestricted.value.hits.len(), 2);
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn cover_window_rpc_rejects_inverted_window() {
|
||||
let (_tmp, cfg) = test_config();
|
||||
let req = CoverWindowRequest {
|
||||
since_ms: 100,
|
||||
until_ms: 50,
|
||||
source_id: None,
|
||||
source_kind: None,
|
||||
limit: None,
|
||||
};
|
||||
let err = cover_window_rpc(&cfg, req).await.unwrap_err();
|
||||
// The guard names both bounds so callers can see the inversion.
|
||||
assert!(err.contains("until_ms"), "got {err}");
|
||||
assert!(err.contains("since_ms"), "got {err}");
|
||||
}
|
||||
|
||||
// ── search_entities_rpc ───────────────────────────────────────────
|
||||
|
||||
#[tokio::test]
|
||||
|
||||
@@ -26,6 +26,7 @@ const NAMESPACE: &str = "memory_tree";
|
||||
pub fn all_controller_schemas() -> Vec<ControllerSchema> {
|
||||
vec![
|
||||
schemas("query_source"),
|
||||
schemas("cover_window"),
|
||||
schemas("search_entities"),
|
||||
schemas("drill_down"),
|
||||
schemas("fetch_leaves"),
|
||||
@@ -40,6 +41,10 @@ pub fn all_registered_controllers() -> Vec<RegisteredController> {
|
||||
schema: schemas("query_source"),
|
||||
handler: handle_query_source,
|
||||
},
|
||||
RegisteredController {
|
||||
schema: schemas("cover_window"),
|
||||
handler: handle_cover_window,
|
||||
},
|
||||
RegisteredController {
|
||||
schema: schemas("search_entities"),
|
||||
handler: handle_search_entities,
|
||||
@@ -133,6 +138,53 @@ pub fn schemas(function: &str) -> ControllerSchema {
|
||||
],
|
||||
outputs: query_response_outputs(),
|
||||
},
|
||||
"cover_window" => ControllerSchema {
|
||||
namespace: NAMESPACE,
|
||||
function: "cover_window",
|
||||
description: "Return the MINIMUM set of nodes covering all memory in a time \
|
||||
window `[since_ms, until_ms]` (epoch-millis). Emits the coarsest summary \
|
||||
whose whole subtree falls inside the window, and raw leaf chunks for \
|
||||
anything not covered by such a summary (boundary content and not-yet-\
|
||||
summarised chunks). Optional `source_id` / `source_kind` scope the result. \
|
||||
Hits are grouped by source and ordered ascending by start time. Use this \
|
||||
for time-bounded recaps (e.g. a last-24h morning brief) instead of \
|
||||
`query_source`, which returns all-time summaries.",
|
||||
inputs: vec![
|
||||
FieldSchema {
|
||||
name: "since_ms",
|
||||
ty: TypeSchema::I64,
|
||||
comment: "Inclusive window start, epoch-milliseconds.",
|
||||
required: true,
|
||||
},
|
||||
FieldSchema {
|
||||
name: "until_ms",
|
||||
ty: TypeSchema::I64,
|
||||
comment: "Inclusive window end, epoch-milliseconds.",
|
||||
required: true,
|
||||
},
|
||||
FieldSchema {
|
||||
name: "source_id",
|
||||
ty: TypeSchema::Option(Box::new(TypeSchema::String)),
|
||||
comment: "Exact source id (e.g. `slack:#eng`, `gmail:abc`).",
|
||||
required: false,
|
||||
},
|
||||
FieldSchema {
|
||||
name: "source_kind",
|
||||
ty: TypeSchema::Option(Box::new(TypeSchema::Enum {
|
||||
variants: vec!["chat", "email", "document"],
|
||||
})),
|
||||
comment: "Source kind filter when no exact id is known.",
|
||||
required: false,
|
||||
},
|
||||
FieldSchema {
|
||||
name: "limit",
|
||||
ty: TypeSchema::Option(Box::new(TypeSchema::U64)),
|
||||
comment: "Max hits (default 200).",
|
||||
required: false,
|
||||
},
|
||||
],
|
||||
outputs: query_response_outputs(),
|
||||
},
|
||||
"search_entities" => ControllerSchema {
|
||||
namespace: NAMESPACE,
|
||||
function: "search_entities",
|
||||
@@ -274,6 +326,14 @@ fn handle_query_source(params: Map<String, Value>) -> ControllerFuture {
|
||||
})
|
||||
}
|
||||
|
||||
fn handle_cover_window(params: Map<String, Value>) -> ControllerFuture {
|
||||
Box::pin(async move {
|
||||
let config = config_rpc::load_config_with_timeout().await?;
|
||||
let req = parse_value::<retrieval_rpc::CoverWindowRequest>(Value::Object(params))?;
|
||||
to_json(retrieval_rpc::cover_window_rpc(&config, req).await?)
|
||||
})
|
||||
}
|
||||
|
||||
fn handle_search_entities(params: Map<String, Value>) -> ControllerFuture {
|
||||
Box::pin(async move {
|
||||
let config = config_rpc::load_config_with_timeout().await?;
|
||||
@@ -318,6 +378,7 @@ mod tests {
|
||||
functions,
|
||||
vec![
|
||||
"query_source",
|
||||
"cover_window",
|
||||
"search_entities",
|
||||
"drill_down",
|
||||
"fetch_leaves",
|
||||
@@ -328,7 +389,7 @@ mod tests {
|
||||
#[test]
|
||||
fn registered_controllers_use_memory_tree_namespace() {
|
||||
let controllers = all_registered_controllers();
|
||||
assert_eq!(controllers.len(), 4);
|
||||
assert_eq!(controllers.len(), 5);
|
||||
assert!(controllers.iter().all(|c| c.schema.namespace == NAMESPACE));
|
||||
}
|
||||
|
||||
|
||||
@@ -140,6 +140,7 @@ pub async fn list_chunks_rpc(
|
||||
until_ms: req.until_ms,
|
||||
limit: req.limit,
|
||||
source_scope: None,
|
||||
exclude_dropped: false,
|
||||
};
|
||||
let rows = tokio::task::spawn_blocking({
|
||||
let config = config.clone();
|
||||
|
||||
@@ -4030,6 +4030,126 @@ async fn json_rpc_memory_tree_end_to_end() {
|
||||
let _ = mock_join.await;
|
||||
}
|
||||
|
||||
/// `openhuman.memory_tree_cover_window` over RPC: ingest a chunk, then assert
|
||||
/// the windowed minimum-cover returns it raw inside the window and nothing
|
||||
/// outside it. One ingested chunk doesn't reach the seal fanout, so this also
|
||||
/// exercises the not-yet-sealed (no Tree row) raw-leaf fallback end to end.
|
||||
#[tokio::test]
|
||||
async fn json_rpc_memory_tree_cover_window_end_to_end() {
|
||||
let _env_lock = json_rpc_e2e_env_lock();
|
||||
let tmp = tempdir().expect("tempdir");
|
||||
let home = tmp.path();
|
||||
let openhuman_home = home.join(".openhuman");
|
||||
|
||||
let _home_guard = EnvVarGuard::set_to_path("HOME", home);
|
||||
let _workspace_guard = EnvVarGuard::unset("OPENHUMAN_WORKSPACE");
|
||||
let _backend_url_guard = EnvVarGuard::unset("BACKEND_URL");
|
||||
let _vite_backend_guard = EnvVarGuard::unset("VITE_BACKEND_URL");
|
||||
// Fall back to the Inert (zero-vector) embedder — no Ollama in CI.
|
||||
let _embed_strict_guard = EnvVarGuard::set("OPENHUMAN_MEMORY_EMBED_STRICT", "false");
|
||||
let _embed_endpoint_guard = EnvVarGuard::set("OPENHUMAN_MEMORY_EMBED_ENDPOINT", "");
|
||||
let _embed_model_guard = EnvVarGuard::set("OPENHUMAN_MEMORY_EMBED_MODEL", "");
|
||||
|
||||
let (mock_addr, mock_join) = serve_on_ephemeral(mock_upstream_router()).await;
|
||||
let mock_origin = format!("http://{}", mock_addr);
|
||||
write_min_config(&openhuman_home, &mock_origin);
|
||||
|
||||
let (rpc_addr, rpc_join) = serve_on_ephemeral(build_core_http_router(false)).await;
|
||||
let rpc_base = format!("http://{}", rpc_addr);
|
||||
tokio::time::sleep(Duration::from_millis(100)).await;
|
||||
|
||||
// Ingest a single document chunk.
|
||||
let ingest = post_json_rpc(
|
||||
&rpc_base,
|
||||
300,
|
||||
"openhuman.memory_tree_ingest",
|
||||
json!({
|
||||
"source_kind": "document",
|
||||
"source_id": "notion:daily-note",
|
||||
"owner": "alice@example.com",
|
||||
"payload": {
|
||||
"provider": "notion",
|
||||
"title": "Daily Note",
|
||||
"body": "Stand-up at 9, shipped the cover_window tool, reviewed the migration plan.",
|
||||
"modified_at": 1_700_000_000_000_i64
|
||||
}
|
||||
}),
|
||||
)
|
||||
.await;
|
||||
let ingest_outer = assert_no_jsonrpc_error(&ingest, "memory_tree_ingest");
|
||||
let ingest_result = ingest_outer.get("result").unwrap_or(ingest_outer);
|
||||
let chunk_ids = ingest_result
|
||||
.get("chunk_ids")
|
||||
.and_then(Value::as_array)
|
||||
.expect("chunk_ids array");
|
||||
assert_eq!(chunk_ids.len(), 1, "expected one ingested chunk");
|
||||
let chunk_id = chunk_ids[0].as_str().expect("chunk id string").to_string();
|
||||
|
||||
// Covering window → the chunk comes back as a raw leaf.
|
||||
let inside = post_json_rpc(
|
||||
&rpc_base,
|
||||
301,
|
||||
"openhuman.memory_tree_cover_window",
|
||||
json!({ "since_ms": 0_i64, "until_ms": 4_000_000_000_000_i64 }),
|
||||
)
|
||||
.await;
|
||||
let inside_outer = assert_no_jsonrpc_error(&inside, "cover_window inside");
|
||||
let inside_result = inside_outer.get("result").unwrap_or(inside_outer);
|
||||
let hits = inside_result
|
||||
.get("hits")
|
||||
.and_then(Value::as_array)
|
||||
.expect("hits array");
|
||||
assert!(
|
||||
hits.iter().any(|h| {
|
||||
h.get("node_id").and_then(Value::as_str) == Some(chunk_id.as_str())
|
||||
&& h.get("node_kind").and_then(Value::as_str) == Some("leaf")
|
||||
}),
|
||||
"covering window must return the ingested chunk as a raw leaf: {inside_result}"
|
||||
);
|
||||
|
||||
// Far-future window (after the chunk) → empty cover.
|
||||
let outside = post_json_rpc(
|
||||
&rpc_base,
|
||||
302,
|
||||
"openhuman.memory_tree_cover_window",
|
||||
json!({ "since_ms": 3_500_000_000_000_i64, "until_ms": 4_000_000_000_000_i64 }),
|
||||
)
|
||||
.await;
|
||||
let outside_outer = assert_no_jsonrpc_error(&outside, "cover_window outside");
|
||||
let outside_result = outside_outer.get("result").unwrap_or(outside_outer);
|
||||
assert_eq!(
|
||||
outside_result
|
||||
.get("hits")
|
||||
.and_then(Value::as_array)
|
||||
.map(|h| h.len()),
|
||||
Some(0),
|
||||
"a window after the chunk must be empty: {outside_result}"
|
||||
);
|
||||
|
||||
// Inverted window is rejected as a JSON-RPC error.
|
||||
let inverted = post_json_rpc(
|
||||
&rpc_base,
|
||||
303,
|
||||
"openhuman.memory_tree_cover_window",
|
||||
json!({ "since_ms": 100_i64, "until_ms": 50_i64 }),
|
||||
)
|
||||
.await;
|
||||
let inverted_err = assert_jsonrpc_error(&inverted, "cover_window inverted");
|
||||
let inverted_msg = inverted_err
|
||||
.get("message")
|
||||
.and_then(Value::as_str)
|
||||
.unwrap_or_default();
|
||||
assert!(
|
||||
inverted_msg.contains("until_ms") && inverted_msg.contains("since_ms"),
|
||||
"expected the inverted-window guard to fire (until_ms < since_ms), got: {inverted_err}"
|
||||
);
|
||||
|
||||
rpc_join.abort();
|
||||
let _ = rpc_join.await;
|
||||
mock_join.abort();
|
||||
let _ = mock_join.await;
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn json_rpc_web_chat_routing_cases_use_expected_backend_models() {
|
||||
run_json_rpc_e2e_on_agent_stack(
|
||||
|
||||
@@ -4385,7 +4385,7 @@ async fn memory_tree_retrieval_rpc_and_schema_wrappers_cover_empty_and_invalid_p
|
||||
openhuman_core::openhuman::memory_tree::retrieval::schemas::all_controller_schemas();
|
||||
let controllers =
|
||||
openhuman_core::openhuman::memory_tree::retrieval::schemas::all_registered_controllers();
|
||||
assert_eq!(schemas.len(), 4);
|
||||
assert_eq!(schemas.len(), 5);
|
||||
assert_eq!(schemas.len(), controllers.len());
|
||||
assert_eq!(
|
||||
openhuman_core::openhuman::memory_tree::retrieval::schemas::schemas("missing").function,
|
||||
|
||||
Reference in New Issue
Block a user