diff --git a/src/openhuman/memory_queue/handlers/mod.rs b/src/openhuman/memory_queue/handlers/mod.rs index 18a75ec37..4fabaaaec 100644 --- a/src/openhuman/memory_queue/handlers/mod.rs +++ b/src/openhuman/memory_queue/handlers/mod.rs @@ -16,7 +16,7 @@ use crate::openhuman::memory::tree_source::get_or_create_source_tree; use crate::openhuman::memory_queue::store; use crate::openhuman::memory_queue::types::{ AppendBufferPayload, AppendTarget, ExtractChunkPayload, FlushStalePayload, Job, JobKind, - JobOutcome, NewJob, NodeRef, ReembedBackfillPayload, SealPayload, + JobOutcome, NewJob, NodeRef, ReembedBackfillPayload, SealDocumentPayload, SealPayload, }; use crate::openhuman::memory_store::chunks::store as chunk_store; use crate::openhuman::memory_store::content::{ @@ -108,7 +108,7 @@ async fn handle_seal_document(config: &Config, job: &Job) -> Result use crate::openhuman::memory::util::redact::redact; use crate::openhuman::memory_tree::tree::seal_document_subtree; - let payload: crate::openhuman::memory_queue::types::SealDocumentPayload = + let payload: SealDocumentPayload = serde_json::from_str(&job.payload_json).context("parse SealDocument payload")?; // doc_id (notion:{conn}:{page}) and tree_scope (notion:{conn}) are diff --git a/src/openhuman/memory_queue/types.rs b/src/openhuman/memory_queue/types.rs index 4eea7e146..dabcd29ca 100644 --- a/src/openhuman/memory_queue/types.rs +++ b/src/openhuman/memory_queue/types.rs @@ -550,6 +550,7 @@ mod tests { fn llm_bound_kinds() { assert!(JobKind::ExtractChunk.is_llm_bound()); assert!(JobKind::Seal.is_llm_bound()); + assert!(JobKind::SealDocument.is_llm_bound()); assert!(JobKind::ReembedBackfill.is_llm_bound()); assert!(!JobKind::AppendBuffer.is_llm_bound()); assert!(!JobKind::FlushStale.is_llm_bound()); diff --git a/src/openhuman/memory_store/content/paths.rs b/src/openhuman/memory_store/content/paths.rs index 355f63433..dfae52dc0 100644 --- a/src/openhuman/memory_store/content/paths.rs +++ b/src/openhuman/memory_store/content/paths.rs @@ -145,12 +145,13 @@ pub fn summary_rel_path_with_layout( }, ) => { let filename = summary_filename(summary_id); + let safe_slug = slugify_source_id(doc_slug); let vfolder = match version_ms { Some(v) => format!("v-{v}"), None => "v-unversioned".to_string(), }; format!( - "{WIKI_PREFIX}/summaries/source-{scope_slug}/docs/{doc_slug}/{vfolder}/L{level}/{filename}.md" + "{WIKI_PREFIX}/summaries/source-{scope_slug}/docs/{safe_slug}/{vfolder}/L{level}/{filename}.md" ) } (SummaryTreeKind::Source, SummaryDiskLayout::Merge) => { @@ -412,7 +413,7 @@ mod tests { ); assert_eq!( p, - "wiki/summaries/source-notion-conn1/docs/notion-conn1-pageA/v-1717500000000/L2/summary-1700000000000-L2-deadbeef.md" + "wiki/summaries/source-notion-conn1/docs/notion-conn1-pagea/v-1717500000000/L2/summary-1700000000000-L2-deadbeef.md" ); } @@ -429,7 +430,7 @@ mod tests { }, ); assert!( - p.contains("/docs/notion-conn1-pageB/v-unversioned/L1/"), + p.contains("/docs/notion-conn1-pageb/v-unversioned/L1/"), "got {p}" ); } diff --git a/src/openhuman/memory_sync/composio/providers/notion/ingest.rs b/src/openhuman/memory_sync/composio/providers/notion/ingest.rs index b744449b9..bc4ecb47a 100644 --- a/src/openhuman/memory_sync/composio/providers/notion/ingest.rs +++ b/src/openhuman/memory_sync/composio/providers/notion/ingest.rs @@ -50,8 +50,11 @@ //! Chunk IDs are content-hashed, so re-ingesting identical content is an //! UPSERT on the same chunk row — no duplicate chunks across syncs. +use std::sync::OnceLock; + use anyhow::Result; use chrono::{DateTime, Utc}; +use regex::Regex; use serde_json::Value; use crate::openhuman::config::Config; @@ -87,6 +90,59 @@ pub(crate) fn notion_source_scope(connection_id: &str) -> String { format!("notion:{connection_id}") } +/// Strip the noise `NOTION_GET_PAGE_MARKDOWN` bakes into the rendered body +/// before it reaches [`render_page_body`] / the ingest pipeline. +/// +/// The Composio markdown export is functional but token-expensive: image +/// links carry S3 *signed* query strings (`?X-Amz-Algorithm=…&X-Amz-Signature=…` +/// — hundreds of tokens of pure noise that also rotate every fetch, defeating +/// content-hash idempotency), inline formatting is wrapped in HTML `` +/// tags that smuggle `discussion-urls=`/`color=` attributes, user @-mentions +/// render as `` tags, and hard line breaks come through as +/// `
`. None of that is useful retrieval signal. This collapses it to clean +/// markdown so the chunked content is the actual prose, headings, and lists. +/// +/// Regexes are compiled once and cached in process-global `OnceLock`s. +/// Conservative by design — it only touches the specific noise patterns above +/// and leaves ordinary markdown (headings, lists, links, fenced code) intact. +fn clean_notion_markdown(md: &str) -> String { + // `![alt](https://host/file.png?X-Amz-...&...)` → `![alt](https://host/file.png)`. + // Match an image link whose URL has a query string, dropping everything + // from the `?` up to the closing paren. The URL itself can't contain `)` + // or whitespace, so `[^)\s?]*` for the path and `[^)]*` for the query is safe. + static IMG_QUERY: OnceLock = OnceLock::new(); + // `inner` → `inner`. Strips the open tag (with any + // attributes: `color=`, `discussion-urls=`, …) and the matching close tag, + // keeping the inner text. Non-greedy attribute match so adjacent spans + // don't get swallowed. + static SPAN_OPEN: OnceLock = OnceLock::new(); + static SPAN_CLOSE: OnceLock = OnceLock::new(); + // `` and `` → removed. + static MENTION: OnceLock = OnceLock::new(); + // `
` / `
` / `
` → newline. + static BR: OnceLock = OnceLock::new(); + // 3+ consecutive newlines (a run of blank lines) → exactly two. + static BLANKS: OnceLock = OnceLock::new(); + + let img_query = + IMG_QUERY.get_or_init(|| Regex::new(r"(!\[[^\]]*\]\([^)\s?]+)\?[^)]*\)").unwrap()); + let span_open = SPAN_OPEN.get_or_init(|| Regex::new(r"(?s)]*>").unwrap()); + let span_close = SPAN_CLOSE.get_or_init(|| Regex::new(r"").unwrap()); + let mention = MENTION.get_or_init(|| { + Regex::new(r"(?s)]*?/>|]*?>.*?").unwrap() + }); + let br = BR.get_or_init(|| Regex::new(r"(?i)").unwrap()); + let blanks = BLANKS.get_or_init(|| Regex::new(r"\n{3,}").unwrap()); + + let s = img_query.replace_all(md, "$1)"); + let s = mention.replace_all(&s, ""); + let s = span_open.replace_all(&s, ""); + let s = span_close.replace_all(&s, ""); + let s = br.replace_all(&s, "\n"); + let s = blanks.replace_all(&s, "\n\n"); + s.trim().to_string() +} + /// Pretty-printed JSON body for one Notion page. We persist the *full* /// Composio response payload (not just the title) so the chunked content /// retains enough context for retrieval — Notion pages don't have a @@ -155,8 +211,11 @@ pub async fn ingest_page_into_memory_tree( // used for read-time latest-wins. let version_ms = modified_at.timestamp_millis(); // Prefer the rendered page body (NOTION_GET_PAGE_MARKDOWN) when present; - // fall back to metadata-only when the caller didn't fetch it. - let body = render_page_body(title, page, markdown_body); + // fall back to metadata-only when the caller didn't fetch it. Strip the + // noise (signed image URLs, ``/`` tags, `
`) from the + // body markdown before rendering — the metadata JSON path is untouched. + let cleaned = markdown_body.map(clean_notion_markdown); + let body = render_page_body(title, page, cleaned.as_deref()); let source_ref = Some(format!("notion://page/{page_id}")); let doc = DocumentInput { @@ -365,6 +424,73 @@ mod tests { assert!(body.contains("```json")); } + /// Signed-image-URL query strings are stripped to bare URLs (keeping the + /// `.png`), so the hundreds of rotating `X-Amz-*` tokens don't pollute the + /// chunk body or defeat content-hash idempotency. + #[test] + fn clean_notion_markdown_strips_image_query() { + let md = "![diagram](https://prod-files.s3.amazonaws.com/a/b/file.png?X-Amz-Algorithm=AWS4-HMAC-SHA256&X-Amz-Credential=AKIA%2F20260604&X-Amz-Signature=deadbeef)"; + let out = clean_notion_markdown(md); + assert_eq!( + out, + "![diagram](https://prod-files.s3.amazonaws.com/a/b/file.png)" + ); + assert!(out.contains(".png"), "the .png extension must be kept"); + assert!(!out.contains("X-Amz-"), "no signed-query noise: {out}"); + } + + /// `text` collapses to its inner text, dropping the tag and + /// any `color=` / `discussion-urls=` attributes. + #[test] + fn clean_notion_markdown_collapses_spans() { + let out = clean_notion_markdown(r#"text"#); + assert_eq!(out, "text"); + + // Attributes like discussion-urls are dropped along with the tags. + let out2 = clean_notion_markdown( + r#"before middle after"#, + ); + assert_eq!(out2, "before middle after"); + } + + /// `` tags are removed entirely (both self-closing and + /// paired forms). + #[test] + fn clean_notion_markdown_removes_mentions() { + let out = clean_notion_markdown(r#"hi there"#); + assert_eq!(out, "hi there"); + assert!(!out.contains("mention-user")); + + let paired = + clean_notion_markdown(r#"@Alice owns it"#); + assert_eq!(paired, "owns it"); + } + + /// `
` / `
` / `
` become newlines. + #[test] + fn clean_notion_markdown_converts_br_to_newline() { + assert_eq!( + clean_notion_markdown("line one
line two"), + "line one\nline two" + ); + assert_eq!(clean_notion_markdown("a
b"), "a\nb"); + assert_eq!(clean_notion_markdown("a
b"), "a\nb"); + } + + /// Plain markdown (no noise patterns) passes through unchanged apart from + /// the trailing-whitespace trim. + #[test] + fn clean_notion_markdown_leaves_plain_markdown_unchanged() { + let md = "# Heading\n\nReal body text with a list:\n- one\n- two\n\n[link](https://example.com/page)"; + assert_eq!(clean_notion_markdown(md), md); + } + + /// 3+ consecutive blank lines collapse to a single blank line (two `\n`). + #[test] + fn clean_notion_markdown_collapses_blank_runs() { + assert_eq!(clean_notion_markdown("a\n\n\n\n\nb"), "a\n\nb"); + } + /// The #2885 regression test. /// /// Before this migration, Notion sync routed through diff --git a/src/openhuman/memory_tree/tree/bucket_seal.rs b/src/openhuman/memory_tree/tree/bucket_seal.rs index ba034a634..b43dd9456 100644 --- a/src/openhuman/memory_tree/tree/bucket_seal.rs +++ b/src/openhuman/memory_tree/tree/bucket_seal.rs @@ -41,6 +41,7 @@ use chrono::{DateTime, Utc}; use rusqlite::Transaction; use crate::openhuman::config::Config; +use crate::openhuman::memory::util::redact::redact; use crate::openhuman::memory_store::chunks::store::with_connection; use crate::openhuman::memory_store::content::{ atomic::stage_summary_with_layout, @@ -1010,15 +1011,15 @@ pub async fn seal_document_subtree( ) -> Result { if chunk_ids.is_empty() { anyhow::bail!( - "[tree::bucket_seal] seal_document_subtree: empty chunk set tree_id={} doc_id={}", + "[tree::bucket_seal] seal_document_subtree: empty chunk set tree_id={} doc_id_hash={}", tree.id, - doc_id + redact(doc_id) ); } log::debug!( - "[tree::bucket_seal] seal_document_subtree tree_id={} doc_id={} version_ms={:?} chunks={}", + "[tree::bucket_seal] seal_document_subtree tree_id={} doc_id_hash={} version_ms={:?} chunks={}", tree.id, - doc_id, + redact(doc_id), version_ms, chunk_ids.len() ); @@ -1060,15 +1061,15 @@ pub async fn seal_document_subtree( let doc_root = doc_root.ok_or_else(|| { anyhow::anyhow!( - "[tree::bucket_seal] seal_document_subtree produced no doc-root tree_id={} doc_id={}", + "[tree::bucket_seal] seal_document_subtree produced no doc-root tree_id={} doc_id_hash={}", tree.id, - doc_id + redact(doc_id) ) })?; log::debug!( - "[tree::bucket_seal] doc-root sealed tree_id={} doc_id={} root_id={} level={}", + "[tree::bucket_seal] doc-root sealed tree_id={} doc_id_hash={} root_id={} level={}", tree.id, - doc_id, + redact(doc_id), doc_root.id, doc_root.level ); @@ -1085,9 +1086,9 @@ pub async fn seal_document_subtree( )?; let merge_sealed = cascade_all_from(config, tree, MERGE_LEVEL_BASE, None, strategy).await?; log::debug!( - "[tree::bucket_seal] merge cascade tree_id={} doc_id={} merge_sealed={}", + "[tree::bucket_seal] merge cascade tree_id={} doc_id_hash={} merge_sealed={}", tree.id, - doc_id, + redact(doc_id), merge_sealed.len() ); @@ -1198,9 +1199,9 @@ async fn seal_explicit_children( // needs compression). let output = if inputs.len() == 1 && inputs[0].token_count <= OUTPUT_TOKEN_BUDGET { log::debug!( - "[tree::bucket_seal] doc-subtree passthrough (1 input, no LLM) tree_id={} doc_id={:?} level={}", + "[tree::bucket_seal] doc-subtree passthrough (1 input, no LLM) tree_id={} doc_id_hash={} level={}", tree.id, - doc_id, + doc_id.map(redact).unwrap_or_default(), level ); let only = &inputs[0]; @@ -1218,8 +1219,8 @@ async fn seal_explicit_children( Ok(o) => o, Err(e) => { log::warn!( - "[tree::bucket_seal] doc-subtree summarise failed tree_id={} doc_id={:?} level={}: {e:#} — fallback", - tree.id, doc_id, level, + "[tree::bucket_seal] doc-subtree summarise failed tree_id={} doc_id_hash={} level={}: {e:#} — fallback", + tree.id, doc_id.map(redact).unwrap_or_default(), level, ); fallback_summary(&inputs, ctx.token_budget) } @@ -1295,6 +1296,13 @@ async fn seal_explicit_children( body: &node.content, }; let content_root = config.memory_tree_content_root(); + if let Err(err) = + crate::openhuman::memory_store::content::obsidian::ensure_obsidian_defaults(&content_root) + { + log::warn!( + "[tree::bucket_seal] ensure_obsidian_defaults failed during doc-subtree seal: {err:#}" + ); + } let doc_slug = doc_id.map(slugify_source_id); let layout = match doc_slug.as_deref() { Some(slug) => SummaryDiskLayout::DocSubtree { @@ -1357,9 +1365,9 @@ async fn seal_explicit_children( })?; log::info!( - "[tree::bucket_seal] doc-subtree sealed tree_id={} doc_id={:?} level={}→{} summary_id={} children={}", + "[tree::bucket_seal] doc-subtree sealed tree_id={} doc_id_hash={} level={}→{} summary_id={} children={}", tree.id, - doc_id, + doc_id.map(redact).unwrap_or_default(), level, target_level, summary_id,