fix(vault): sync writes to memory_tree, not legacy UnifiedMemory (#2705)

src/openhuman/vault/ops.rs

@@ -6,6 +6,8 @@ use uuid::Uuid;
 
 use crate::openhuman::config::Config;
 use crate::openhuman::memory::ops::{clear_namespace, ClearNamespaceParams};
+use crate::openhuman::memory_store::chunks::store::delete_chunks_by_source_prefix;
+use crate::openhuman::memory_store::chunks::types::SourceKind;
 use crate::rpc::RpcOutcome;
 
 use super::state;
@@ -113,23 +115,68 @@ pub async fn vault_remove(
     log::debug!("[vault] remove: id={id} removed={removed} purge_memory={purge_memory}");
 
     let mut purged = false;
+    let mut memory_tree_chunks_deleted: usize = 0;
     if removed && purge_memory {
         if let Some(v) = vault {
+            // Memory-tree cleanup is the canonical path post-#2705: vault
+            // sync writes to `mem_tree_chunks` / `mem_tree_ingested_sources`
+            // keyed by `vault:{id}:{rel_path}`. A prefix delete with
+            // `vault:{id}:` catches every per-file row for this vault. The
+            // companion `clear_namespace` call below still drains any
+            // pre-#2705 ledger rows that landed in the legacy
+            // `memory_docs` table during the migration window.
+            let cfg_for_blocking = config.clone();
+            let prefix = format!("vault:{}:", v.id);
+            let tree_result = tokio::task::spawn_blocking(move || {
+                delete_chunks_by_source_prefix(&cfg_for_blocking, SourceKind::Document, &prefix)
+            })
+            .await;
+            match tree_result {
+                Ok(Ok(removed_chunks)) => {
+                    memory_tree_chunks_deleted = removed_chunks;
+                    log::debug!(
+                        "[vault] remove: id={id} memory_tree_chunks_deleted={removed_chunks}"
+                    );
+                }
+                Ok(Err(err)) => {
+                    log::warn!("[vault] remove: id={id} memory_tree_purge_failed err={err}");
+                    return Ok(RpcOutcome::single_log(
+                        serde_json::json!({
+                            "vault_id": id,
+                            "removed": removed,
+                            "purged": false,
+                            "purge_error": format!("memory_tree purge failed: {err}"),
+                        }),
+                        format!("vault removed with purge error: {id}"),
+                    ));
+                }
+                Err(join_err) => {
+                    log::warn!(
+                        "[vault] remove: id={id} memory_tree_purge_join_failed err={join_err}"
+                    );
+                    return Ok(RpcOutcome::single_log(
+                        serde_json::json!({
+                            "vault_id": id,
+                            "removed": removed,
+                            "purged": false,
+                            "purge_error": format!("memory_tree purge join error: {join_err}"),
+                        }),
+                        format!("vault removed with purge error: {id}"),
+                    ));
+                }
+            }
+
+            // Best-effort legacy UnifiedMemory purge for pre-#2705 ledger
+            // rows whose chunks still live in `memory_docs`. Failure here
+            // doesn't undo the canonical memory_tree cleanup above.
             if let Err(err) = clear_namespace(ClearNamespaceParams {
                 namespace: v.namespace.clone(),
             })
             .await
             {
-                log::warn!("[vault] remove: id={id} purge_namespace_failed err={err}");
-                return Ok(RpcOutcome::single_log(
-                    serde_json::json!({
-                        "vault_id": id,
-                        "removed": removed,
-                        "purged": false,
-                        "purge_error": err,
-                    }),
-                    format!("vault removed with purge error: {id}"),
-                ));
+                log::debug!(
+                    "[vault] remove: id={id} legacy_clear_namespace_failed (best-effort) err={err}"
+                );
             }
             purged = true;
         }
@@ -140,6 +187,7 @@ pub async fn vault_remove(
             "vault_id": id,
             "removed": removed,
             "purged": purged,
+            "memory_tree_chunks_deleted": memory_tree_chunks_deleted,
         }),
         format!("vault removed: {id}"),
     ))