fix(memory_tree): gate ingest on source_id so summariser tree never sees a source twice (#1353)

src/openhuman/memory/tree/ingest.rs

@@ -21,6 +21,8 @@ use crate::openhuman::memory::tree::content_store;
 use crate::openhuman::memory::tree::jobs::{self, ExtractChunkPayload, NewJob};
 use crate::openhuman::memory::tree::score::{self, ScoreResult, ScoringConfig};
 use crate::openhuman::memory::tree::store;
+use crate::openhuman::memory::tree::types::SourceKind;
+use crate::openhuman::memory::tree::util::redact::redact;
 
 /// Outcome of one ingest call.
 #[derive(Clone, Debug, Serialize, Deserialize)]
@@ -33,6 +35,11 @@ pub struct IngestResult {
     pub chunks_dropped: usize,
     /// IDs of all chunks written and queued.
     pub chunk_ids: Vec<String>,
+    /// True when this ingest was a no-op because `(source_kind, source_id)`
+    /// had already been ingested. Memory items are append-only — the
+    /// summariser tree must not see the same source twice.
+    #[serde(default)]
+    pub already_ingested: bool,
 }
 
 impl IngestResult {
@@ -42,6 +49,17 @@ impl IngestResult {
             chunks_written: 0,
             chunks_dropped: 0,
             chunk_ids: Vec::new(),
+            already_ingested: false,
+        }
+    }
+
+    fn already_ingested(source_id: &str) -> Self {
+        Self {
+            source_id: source_id.to_string(),
+            chunks_written: 0,
+            chunks_dropped: 0,
+            chunk_ids: Vec::new(),
+            already_ingested: true,
         }
     }
 }
@@ -55,6 +73,12 @@ pub async fn ingest_chat(
     tags: Vec<String>,
     batch: ChatBatch,
 ) -> Result<IngestResult> {
+    // No source-level gate for chat: a chat `source_id` (e.g.
+    // `slack:{connection_id}`) is a stream identifier — many batches /
+    // buckets accumulate into the same source tree over time. The gate
+    // would make every bucket after the first a no-op. Chunk-level
+    // idempotency (`chunk_id` includes content) still prevents true
+    // replay duplicates from reaching the summariser.
     let canonical =
         match chat::canonicalise(source_id, owner, &tags, batch).map_err(anyhow::Error::msg)? {
             Some(c) => c,
@@ -72,6 +96,11 @@ pub async fn ingest_email(
     tags: Vec<String>,
     thread: EmailThread,
 ) -> Result<IngestResult> {
+    // No source-level gate for email: gmail per-participant ingest
+    // groups many threads under one `source_id` (e.g.
+    // `gmail:{participants_hash}`) and appends as new threads arrive.
+    // The gate would block all but the first thread. Chunk-level
+    // idempotency still protects against true replays.
     let canonical =
         match email::canonicalise(source_id, owner, &tags, thread).map_err(anyhow::Error::msg)? {
             Some(c) => c,
@@ -89,6 +118,13 @@ pub async fn ingest_document(
     tags: Vec<String>,
     doc: DocumentInput,
 ) -> Result<IngestResult> {
+    if already_ingested(config, SourceKind::Document, source_id).await? {
+        log::debug!(
+            "[memory_tree::ingest] skip ingest_document — source_id_hash={} already ingested",
+            redact(source_id)
+        );
+        return Ok(IngestResult::already_ingested(source_id));
+    }
     let canonical =
         match document::canonicalise(source_id, owner, &tags, doc).map_err(anyhow::Error::msg)? {
             Some(c) => c,
@@ -97,11 +133,27 @@ pub async fn ingest_document(
     persist(config, source_id, canonical).await
 }
 
+/// Best-effort pre-canonicalisation check. The transactional claim inside
+/// [`persist`] is what actually serialises concurrent ingests; this lookup
+/// just spares the canonicaliser when we already know the source is a dup.
+async fn already_ingested(
+    config: &Config,
+    source_kind: SourceKind,
+    source_id: &str,
+) -> Result<bool> {
+    let cfg = config.clone();
+    let sid = source_id.to_string();
+    tokio::task::spawn_blocking(move || store::is_source_ingested(&cfg, source_kind, &sid))
+        .await
+        .map_err(|e| anyhow::anyhow!("already_ingested join error: {e}"))?
+}
+
 async fn persist(
     config: &Config,
     source_id: &str,
     canonical: CanonicalisedSource,
 ) -> Result<IngestResult> {
+    let source_kind_for_store = canonical.metadata.source_kind;
     let input = ChunkerInput {
         source_kind: canonical.metadata.source_kind,
         source_id: source_id.to_string(),
@@ -140,11 +192,45 @@ async fn persist(
     let config_owned = config.clone();
     let staged_for_store = staged.clone();
     let results_for_store = all_results.clone();
-    let written = tokio::task::spawn_blocking(move || -> Result<usize> {
+    let source_id_for_store = source_id.to_string();
+    let written = tokio::task::spawn_blocking(move || -> Result<Option<usize>> {
         use std::collections::{HashMap, HashSet};
         store::with_connection(&config_owned, |conn| {
             let tx = conn.unchecked_transaction()?;
 
+            // Authoritative source-level gate (documents only).
+            //
+            // For documents, `source_id` identifies a single immutable
+            // file (one notion page, one drive doc). `is_source_ingested`
+            // above is a best-effort fast-path; this row insert is what
+            // actually serialises concurrent ingests of the same
+            // document and prevents the same content from flowing
+            // through extract → admit → buffer → seal twice.
+            //
+            // Chat and email don't get this gate: their `source_id`
+            // is a *stream* identifier (e.g. slack workspace, gmail
+            // participant group) under which many batches / threads
+            // accumulate over time. The chunk-level idempotency in
+            // the rest of this transaction is enough to swallow
+            // genuine replays without blocking legitimate appends.
+            if source_kind_for_store == SourceKind::Document {
+                let now_ms = chrono::Utc::now().timestamp_millis();
+                let claimed = store::claim_source_ingest_tx(
+                    &tx,
+                    source_kind_for_store,
+                    &source_id_for_store,
+                    now_ms,
+                )?;
+                if !claimed {
+                    log::debug!(
+                        "[memory_tree::ingest] persist gate: document already ingested source_id_hash={}",
+                        redact(&source_id_for_store)
+                    );
+                    // Drop the (empty) transaction implicitly; nothing to commit.
+                    return Ok(None);
+                }
+            }
+
             // Read each chunk's CURRENT lifecycle BEFORE the upsert. This
             // is the "did this chunk exist before this batch" snapshot,
             // because `upsert_staged_chunks_tx` will either preserve the
@@ -205,12 +291,21 @@ async fn persist(
                 let _ = jobs::enqueue_tx(&tx, &extract)?;
             }
             tx.commit()?;
-            Ok(n)
+            Ok(Some(n))
         })
     })
     .await
     .map_err(|e| anyhow::anyhow!("persist join error: {e}"))??;
 
+    let written = match written {
+        Some(n) => n,
+        None => {
+            // Lost the race against a concurrent ingest of the same source —
+            // the other writer claimed the row first. No work was committed.
+            return Ok(IngestResult::already_ingested(source_id));
+        }
+    };
+
     jobs::wake_workers();
 
     Ok(IngestResult {
@@ -218,6 +313,7 @@ async fn persist(
         chunks_written: written,
         chunks_dropped: dropped,
         chunk_ids: staged.iter().map(|s| s.chunk.id.clone()).collect(),
+        already_ingested: false,
     })
 }
 
@@ -345,6 +441,41 @@ mod tests {
         assert_eq!(count_scores(&cfg).unwrap(), 0);
     }
 
+    #[tokio::test]
+    async fn second_ingest_document_with_same_source_id_is_short_circuited() {
+        let (_tmp, cfg) = test_config();
+        let doc = DocumentInput {
+            provider: "notion".into(),
+            title: "Launch plan".into(),
+            body: "Phoenix ships Friday after staging review. alice@example.com owns this.".into(),
+            modified_at: Utc.timestamp_millis_opt(1_700_000_000_000).unwrap(),
+            source_ref: Some("notion://page/abc".into()),
+        };
+        let first = ingest_document(&cfg, "notion:abc", "alice", vec![], doc.clone())
+            .await
+            .unwrap();
+        assert!(!first.already_ingested);
+        assert!(first.chunks_written >= 1);
+
+        // Even with completely different content under the same source_id,
+        // the second ingest must not write anything: documents are
+        // append-only and the source_id is the dedup key.
+        let mutated = DocumentInput {
+            body: "totally different content that should NOT make it into the tree".into(),
+            ..doc
+        };
+        let second = ingest_document(&cfg, "notion:abc", "alice", vec![], mutated)
+            .await
+            .unwrap();
+        assert!(second.already_ingested);
+        assert_eq!(second.chunks_written, 0);
+        assert!(second.chunk_ids.is_empty());
+
+        drain_until_idle(&cfg).await.unwrap();
+        // Only the first ingest's chunks made it into the store.
+        assert_eq!(count_chunks(&cfg).unwrap(), first.chunks_written as u64);
+    }
+
     #[tokio::test]
     async fn re_ingest_is_idempotent_on_chunks_and_scores() {
         let (_tmp, cfg) = test_config();

src/openhuman/memory/tree/store.rs

@@ -215,6 +215,19 @@ CREATE INDEX IF NOT EXISTS idx_mem_tree_jobs_kind
 CREATE UNIQUE INDEX IF NOT EXISTS idx_mem_tree_jobs_dedupe_active
     ON mem_tree_jobs(dedupe_key)
     WHERE dedupe_key IS NOT NULL AND status IN ('ready', 'running');
+
+-- Source-level ingest gate. Memory items (documents, chat batches, email
+-- threads) are append-only — once a `(source_kind, source_id)` is ingested
+-- it must not be re-ingested, otherwise its chunks flow back through
+-- extract → admit → buffer → seal and end up duplicated in the summariser
+-- tree. The first ingest claims the row; subsequent ingest_* calls for the
+-- same key short-circuit before canonicalisation.
+CREATE TABLE IF NOT EXISTS mem_tree_ingested_sources (
+    source_kind            TEXT NOT NULL,
+    source_id              TEXT NOT NULL,
+    ingested_at_ms         INTEGER NOT NULL,
+    PRIMARY KEY (source_kind, source_id)
+);
 ";
 
 /// Upsert a batch of chunks atomically.
@@ -532,6 +545,49 @@ fn set_chunk_lifecycle_status_conn(conn: &Connection, chunk_id: &str, status: &s
     Ok(())
 }
 
+/// Best-effort, non-transactional check used by `ingest_*` to skip
+/// canonicalisation when a source has already been ingested. The
+/// authoritative gate is [`claim_source_ingest_tx`] inside the persist
+/// transaction — this lookup just avoids burning canonicaliser work on
+/// the obvious dup case.
+pub fn is_source_ingested(
+    config: &Config,
+    source_kind: SourceKind,
+    source_id: &str,
+) -> Result<bool> {
+    with_connection(config, |conn| {
+        let n: i64 = conn.query_row(
+            "SELECT COUNT(*) FROM mem_tree_ingested_sources \
+             WHERE source_kind = ?1 AND source_id = ?2",
+            params![source_kind.as_str(), source_id],
+            |r| r.get(0),
+        )?;
+        Ok(n > 0)
+    })
+}
+
+/// Atomically claim `(source_kind, source_id)` for ingestion. Returns
+/// `true` if the row was newly inserted (caller should proceed with the
+/// rest of the persist transaction); `false` if a previous ingest already
+/// claimed this source (caller must roll back / skip).
+///
+/// Lives inside the same transaction as the chunk + job writes so two
+/// concurrent ingests of the same source can't both pass the gate.
+pub(crate) fn claim_source_ingest_tx(
+    tx: &Transaction<'_>,
+    source_kind: SourceKind,
+    source_id: &str,
+    now_ms: i64,
+) -> Result<bool> {
+    let inserted = tx.execute(
+        "INSERT OR IGNORE INTO mem_tree_ingested_sources \
+            (source_kind, source_id, ingested_at_ms) \
+         VALUES (?1, ?2, ?3)",
+        params![source_kind.as_str(), source_id, now_ms],
+    )?;
+    Ok(inserted > 0)
+}
+
 fn row_to_chunk(row: &rusqlite::Row<'_>) -> rusqlite::Result<Chunk> {
     let id: String = row.get(0)?;
     let source_kind_s: String = row.get(1)?;