fix(ingest): persist staged content pointers

src/memory/ingest/pipeline.rs

@@ -17,7 +17,7 @@ use chrono::Utc;
 
 use crate::memory::chunks::{
     self, chunk_markdown, claim_source_ingest_tx, delete_source_ingest, get_chunk_lifecycle_status,
-    is_source_ingested, set_chunk_lifecycle_status, set_chunk_raw_refs, upsert_chunks,
+    is_source_ingested, set_chunk_lifecycle_status, set_chunk_raw_refs, upsert_staged_chunks_tx,
     with_connection, ChunkerInput, ChunkerOptions, SourceKind, CHUNK_STATUS_PENDING_EXTRACTION,
 };
 use crate::memory::config::MemoryConfig;
@@ -146,7 +146,7 @@ async fn persist_score_enqueue(
 ) -> Result<IngestSummary> {
     // 3. Write each chunk body to the content store (atomic write + sha256).
     let content_root = chunks::content_root(config);
-    content::stage_chunks(&content_root, &chunks)
+    let staged = content::stage_chunks(&content_root, &chunks)
         .map_err(|e| anyhow!("stage_chunks failed: {e}"))?;
 
     // 4. Snapshot each chunk's CURRENT lifecycle BEFORE the upsert. A chunk that
@@ -159,7 +159,12 @@ async fn persist_score_enqueue(
     }
 
     // 5. Persist chunk rows (idempotent on deterministic chunk id).
-    let chunks_written = upsert_chunks(config, &chunks)?;
+    let chunks_written = with_connection(config, |connection| {
+        let transaction = connection.unchecked_transaction()?;
+        let count = upsert_staged_chunks_tx(&transaction, &staged)?;
+        transaction.commit()?;
+        Ok(count)
+    })?;
 
     // 5b. Raw-archive-backed bodies: attach refs so a worker can resolve them.
     if let Some(refs) = opts.raw_refs.as_ref() {

src/memory/ingest/pipeline_tests.rs

@@ -6,8 +6,8 @@ use tempfile::TempDir;
 
 use super::{ingest_chat, ingest_document, ingest_document_versioned, ingest_email_with_raw_refs};
 use crate::memory::chunks::{
-    count_chunks, get_chunk_lifecycle_status, is_source_ingested, SourceKind,
-    CHUNK_STATUS_PENDING_EXTRACTION,
+    count_chunks, get_chunk_content_pointers, get_chunk_lifecycle_status, is_source_ingested,
+    SourceKind, CHUNK_STATUS_PENDING_EXTRACTION,
 };
 use crate::memory::chunks::{get_chunk_raw_refs, RawRef};
 use crate::memory::config::MemoryConfig;
@@ -99,6 +99,11 @@ async fn ingest_chat_writes_chunks_and_enqueues_extract_jobs() {
     assert!(out.chunks_written >= 1);
     assert_eq!(count_chunks(&cfg).unwrap(), out.chunks_written as u64);
     assert_eq!(out.chunk_ids.len(), out.chunks_written);
+    let pointers = get_chunk_content_pointers(&cfg, &out.chunk_ids[0])
+        .unwrap()
+        .unwrap();
+    assert!(!pointers.0.is_empty());
+    assert!(!pointers.1.is_empty());
 
     // Every scheduled chunk got an extract job and is parked at pending.
     assert_eq!(out.extract_jobs_enqueued, out.chunk_ids.len());

src/memory/store/kv.rs

@@ -98,12 +98,15 @@ impl KvStore {
     /// characters collapse to `_` so `"team alpha/#1"` and `"team_alpha/_1"`
     /// address the same bucket.
     fn sanitize_namespace(namespace: &str) -> String {
-        namespace
+        let trimmed = namespace.trim();
+        if trimmed.is_empty() {
+            return crate::memory::types::GLOBAL_NAMESPACE.to_string();
+        }
+        trimmed
             .chars()
             .map(|c| match c {
-                c if c.is_whitespace() => '_',
-                '#' => '_',
-                other => other,
+                c if c.is_ascii_alphanumeric() || matches!(c, '-' | '_' | '/') => c,
+                _ => '_',
             })
             .collect()
     }

src/memory/sync/composio/client.rs

@@ -221,13 +221,24 @@ impl ComposioClient {
             let _ = response.bytes().await;
             anyhow::bail!("Composio proxy request failed with HTTP {status}");
         }
-        response
+        let raw: serde_json::Value = response
             .json()
             .await
-            .map_err(|error| anyhow::anyhow!("Composio proxy response decode failed: {error}"))
+            .map_err(|error| anyhow::anyhow!("Composio proxy response decode failed: {error}"))?;
+        decode_proxy_response(raw)
     }
 }
 
+fn decode_proxy_response(raw: serde_json::Value) -> anyhow::Result<ExecuteResponse> {
+    let payload = if raw.get("successful").is_some() {
+        raw
+    } else {
+        raw.get("data").cloned().unwrap_or(raw)
+    };
+    serde_json::from_value(payload)
+        .map_err(|error| anyhow::anyhow!("Composio proxy response decode failed: {error}"))
+}
+
 fn retryable_provider_error(error: Option<&str>) -> bool {
     error.is_some_and(|error| {
         let lower = error.to_ascii_lowercase();
@@ -259,3 +270,36 @@ async fn decode_response(
         .await
         .map_err(|error| anyhow::anyhow!("Composio {mode} response decode failed: {error}"))
 }
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+
+    #[test]
+    fn proxied_backend_envelope_decodes_provider_response() {
+        let response = decode_proxy_response(serde_json::json!({
+            "success": true,
+            "data": {
+                "successful": true,
+                "data": {"messages": [{"messageId": "message-1"}]},
+                "error": null
+            }
+        }))
+        .unwrap();
+
+        assert!(response.successful);
+        assert_eq!(response.data["messages"][0]["messageId"], "message-1");
+    }
+
+    #[test]
+    fn flat_proxy_response_remains_supported() {
+        let response = decode_proxy_response(serde_json::json!({
+            "successful": true,
+            "data": {"items": [1]}
+        }))
+        .unwrap();
+
+        assert!(response.successful);
+        assert_eq!(response.data["items"], serde_json::json!([1]));
+    }
+}

src/memory/sync/composio/gmail.rs

@@ -82,6 +82,9 @@ impl IncrementalSource for GmailSyncPipeline {
     fn max_pages(&self) -> usize {
         self.max_pages
     }
+    fn stop_on_empty_pending(&self) -> bool {
+        true
+    }
 
     fn server_side_depth(&self) -> bool {
         true

src/memory/sync/composio/orchestrator.rs

@@ -65,6 +65,9 @@ pub trait IncrementalSource: Send + Sync {
     fn retain_dedup_keys(&self) -> bool {
         true
     }
+    fn stop_on_empty_pending(&self) -> bool {
+        false
+    }
     fn server_side_depth(&self) -> bool {
         false
     }
@@ -271,6 +274,7 @@ async fn run_pages(
 
             let fetched = source.extract_page(&response.data, page_token.as_deref());
             let mut reached_cursor_boundary = false;
+            let mut saw_unsynced_item = false;
             for raw in fetched.items {
                 if config
                     .sync
@@ -291,6 +295,7 @@ async fn run_pages(
                 if state.is_synced(&dedup_key) {
                     continue;
                 }
+                saw_unsynced_item = true;
                 let sort_cursor = source.sort_cursor(&raw);
                 if sort_cursor
                     .as_deref()
@@ -376,6 +381,15 @@ async fn run_pages(
             }
 
             page_token = fetched.next;
+            if source.stop_on_empty_pending() && !saw_unsynced_item {
+                tracing::debug!(
+                    toolkit = source.toolkit(),
+                    connection_id,
+                    scope = %scope.label,
+                    "[sync:orchestrator] stopping after all-deduplicated page"
+                );
+                break;
+            }
             if reached_cursor_boundary {
                 break;
             }