feat(memory): kind-aware raw layout, _source.md mirror, source/<slug> tag (tinyhumansai#1351)

Author: senamakel

src/openhuman/composio/providers/gmail/ingest.rs

@@ -25,9 +25,8 @@ use crate::openhuman::memory::tree::canonicalize::email::{EmailMessage, EmailThr
 use crate::openhuman::memory::tree::canonicalize::email_clean::{
     extract_email, parse_message_date,
 };
-use crate::openhuman::memory::tree::content_store::paths::slugify_source_id;
 use crate::openhuman::memory::tree::content_store::raw::{
-    self as raw_store, slug_account_email, RawItem,
+    self as raw_store, raw_rel_path, slug_account_email, RawItem, RawKind,
 };
 use crate::openhuman::memory::tree::ingest::{ingest_email, IngestResult};
 use crate::openhuman::memory::tree::store::{set_chunk_raw_refs, RawRef};
@@ -222,7 +221,7 @@ fn parse_address_list(v: Option<&Value>) -> Vec<String> {
 ///
 /// In addition to the chunked content_store output, we mirror every
 /// admitted message as a verbatim `.md` under
-/// `<content_root>/raw/<source_slug>/<created_at_ms>_<message_id>.md`.
+/// `<content_root>/raw/<source_slug>/emails/<created_at_ms>_<message_id>.md`.
 /// Useful for debugging, Obsidian browsing, and as a stable archive
 /// independent of the chunker / summariser.
 ///
@@ -334,7 +333,6 @@ async fn ingest_per_message(
     owner: &str,
     page_messages: &[Value],
 ) -> usize {
-    let source_slug = slugify_source_id(source_id);
     let mut total_chunks = 0usize;
     for raw in page_messages {
         let id = raw
@@ -349,11 +347,11 @@ async fn ingest_per_message(
             continue;
         };
 
-        let raw_path = format!(
-            "raw/{}/{}_{}.md",
-            source_slug,
+        let raw_path = raw_rel_path(
+            source_id,
+            RawKind::Email,
             sent_at.timestamp_millis(),
-            sanitize_uid_for_path(msg_id)
+            msg_id,
         );
 
         let thread_subject = pick_thread_subject(std::slice::from_ref(&message));
@@ -393,27 +391,9 @@ async fn ingest_per_message(
     total_chunks
 }
 
-/// Same character map the raw-archive writer uses for filenames.
-/// Mirrors `raw_store::write_raw_items::sanitize_uid` but local so a
-/// future rule change on either side stays decoupled.
-fn sanitize_uid_for_path(uid: &str) -> String {
-    let cleaned: String = uid
-        .chars()
-        .map(|c| match c {
-            '\\' | '/' | ':' | '*' | '?' | '"' | '<' | '>' | '|' | ' ' => '-',
-            other => other,
-        })
-        .collect();
-    if cleaned.is_empty() {
-        "unknown".into()
-    } else {
-        cleaned
-    }
-}
-
 /// Mirror a page of raw Gmail messages into the on-disk raw archive.
 ///
-/// Files land under `<content_root>/raw/<source_slug>/<ts_ms>_<msg_id>.md`.
+/// Files land under `<content_root>/raw/<source_slug>/emails/<ts_ms>_<msg_id>.md`.
 /// We write the **backend-produced markdown verbatim** — the
 /// `markdown` field on each message is the per-message slice of the
 /// response-level `markdownFormatted`, pinned by
@@ -493,6 +473,7 @@ fn write_raw_archive(config: &Config, source_id: &str, page: &[Value]) -> Result
             uid: id,
             created_at_ms: *ts,
             markdown: md.as_str(),
+            kind: RawKind::Email,
         })
         .collect();
 

src/openhuman/memory/tree/content_store/compose.rs

@@ -38,9 +38,61 @@
 
 use chrono::{DateTime, Utc};
 
-use crate::openhuman::memory::tree::content_store::paths::{sanitize_filename, SummaryTreeKind};
+use crate::openhuman::memory::tree::content_store::paths::{
+    sanitize_filename, slugify_source_id, SummaryTreeKind,
+};
 use crate::openhuman::memory::tree::types::{Chunk, SourceKind};
 
+/// Build the canonical Obsidian `source/<slug>` tag for a given
+/// `source_id`. Used to seed the `tags:` block on every chunk and
+/// every source-tree summary so the Obsidian graph view can filter by
+/// source.
+///
+/// Slug rules match `slugify_source_id` (lowercase ASCII, `-` separators,
+/// alphanumerics + `_` preserved) so the tag matches the on-disk
+/// `raw/<slug>/...` directory name byte-for-byte.
+pub fn source_tag(source_id: &str) -> String {
+    format!("source/{}", slugify_source_id(source_id))
+}
+
+/// Prepend the source tag to `tags`, dedup, and return the new list.
+/// Order is preserved otherwise — `source/...` always comes first so
+/// it shows up at the top of the YAML block.
+pub fn with_source_tag(source_id: &str, tags: &[String]) -> Vec<String> {
+    let st = source_tag(source_id);
+    let mut out = Vec::with_capacity(tags.len() + 1);
+    out.push(st.clone());
+    for t in tags {
+        if t != &st {
+            out.push(t.clone());
+        }
+    }
+    out
+}
+
+/// Parse the value of a top-level YAML scalar field (e.g. `source_id`,
+/// `tree_scope`, `tree_kind`) from a frontmatter string. Strips
+/// surrounding double-quotes if present so the returned slice matches
+/// what the original composer passed in. Returns `None` if the key is
+/// not present at the top level of the frontmatter.
+pub fn scan_fm_field<'a>(fm: &'a str, key: &str) -> Option<String> {
+    let prefix = format!("{key}: ");
+    for raw in fm.lines() {
+        // Skip indented lines (those are list items / nested mappings).
+        if raw.starts_with(' ') || raw.starts_with('\t') {
+            continue;
+        }
+        if let Some(rest) = raw.strip_prefix(&prefix) {
+            let trimmed = rest.trim();
+            if let Some(inner) = trimmed.strip_prefix('"').and_then(|s| s.strip_suffix('"')) {
+                return Some(inner.replace("\\\"", "\"").replace("\\\\", "\\"));
+            }
+            return Some(trimmed.to_string());
+        }
+    }
+    None
+}
+
 /// Compose the full file content (front-matter + body) for `chunk`.
 ///
 /// Returns `(full_file_bytes, body_bytes)`. The caller writes `full_file_bytes`
@@ -78,13 +130,13 @@ fn build_front_matter(chunk: &Chunk) -> Vec<u8> {
         fm.push_str(&format!("source_ref: {}\n", yaml_scalar(&sr.value)));
     }
 
-    if meta.tags.is_empty() {
-        fm.push_str("tags: []\n");
-    } else {
-        fm.push_str("tags:\n");
-        for tag in &meta.tags {
-            fm.push_str(&format!("  - {}\n", yaml_scalar(tag)));
-        }
+    // Always seed the source tag so the Obsidian graph filter can pick
+    // up `source/<slug>` for every chunk regardless of what the
+    // ingest-side tag list contained.
+    let seeded_tags = with_source_tag(&meta.source_id, &meta.tags);
+    fm.push_str("tags:\n");
+    for tag in &seeded_tags {
+        fm.push_str(&format!("  - {}\n", yaml_scalar(tag)));
     }
 
     // Email-specific fields: participants list + Obsidian alias.
@@ -369,7 +421,16 @@ fn build_summary_front_matter(r: &SummaryComposeInput<'_>) -> String {
     fm.push_str("aliases:\n");
     fm.push_str(&format!("  - {}\n", yaml_scalar(&alias)));
 
-    fm.push_str("tags: []\n");
+    // Source-tree summaries get a `source/<slug>` seed tag for graph
+    // filtering. Global / topic trees aggregate across sources, so the
+    // `source/...` tag has no single value there — leave them untagged
+    // at compose time (LLM extraction adds entity tags later).
+    if matches!(r.tree_kind, SummaryTreeKind::Source) {
+        fm.push_str("tags:\n");
+        fm.push_str(&format!("  - {}\n", yaml_scalar(&source_tag(r.tree_scope))));
+    } else {
+        fm.push_str("tags: []\n");
+    }
     fm.push_str("---\n");
     fm
 }
@@ -790,7 +851,10 @@ mod tests {
             fm.contains("  - \"[[child-2]]\""),
             "must list child ids as Obsidian wikilinks; got:\n{fm}"
         );
-        assert!(fm.contains("tags: []"), "must start with empty tags");
+        assert!(
+            fm.contains("  - source/"),
+            "source-tree summary must seed source tag; got:\n{fm}"
+        );
         // aliases must mention the scope
         assert!(fm.contains("aliases:"), "must have aliases");
         assert!(