fix(memory): replace chunks when re-indexing sources (#675)

rust/crates/openjarvis-python/src/storage.rs

@@ -33,6 +33,33 @@ impl PySQLiteMemory {
             .map_err(|e| PyErr::new::<pyo3::exceptions::PyRuntimeError, _>(e.to_string()))
     }
 
+    fn replace_source(
+        &self,
+        source: &str,
+        documents: Vec<(String, Option<String>)>,
+    ) -> PyResult<Vec<String>> {
+        let parsed_documents = documents
+            .into_iter()
+            .map(|(content, metadata)| {
+                let metadata = metadata
+                    .map(|value| serde_json::from_str(&value))
+                    .transpose()
+                    .map_err(|e| {
+                        PyErr::new::<pyo3::exceptions::PyValueError, _>(e.to_string())
+                    })?;
+                Ok((content, metadata))
+            })
+            .collect::<PyResult<Vec<_>>>()?;
+        let document_refs = parsed_documents
+            .iter()
+            .map(|(content, metadata)| (content.as_str(), metadata.as_ref()))
+            .collect::<Vec<_>>();
+
+        self.inner
+            .replace_source(source, &document_refs)
+            .map_err(|e| PyErr::new::<pyo3::exceptions::PyRuntimeError, _>(e.to_string()))
+    }
+
     #[pyo3(signature = (query, top_k=5))]
     fn retrieve(&self, query: &str, top_k: usize) -> PyResult<String> {
         let results = self

rust/crates/openjarvis-tools/src/storage/sqlite.rs

@@ -94,6 +94,57 @@ impl SQLiteMemory {
     pub fn in_memory() -> Result<Self, OpenJarvisError> {
         Self::new(Path::new(":memory:"))
     }
+
+    /// Atomically replace every document for *source* with *documents*.
+    pub fn replace_source(
+        &self,
+        source: &str,
+        documents: &[(&str, Option<&Value>)],
+    ) -> Result<Vec<String>, OpenJarvisError> {
+        let mut conn = self.conn.lock();
+        let tx = conn.transaction().map_err(|e| {
+            OpenJarvisError::Io(std::io::Error::other(e.to_string()))
+        })?;
+
+        tx.execute(
+            "DELETE FROM documents_fts
+             WHERE rowid IN (SELECT rowid FROM documents WHERE source = ?1)",
+            rusqlite::params![source],
+        )
+        .map_err(|e| OpenJarvisError::Io(std::io::Error::other(e.to_string())))?;
+        tx.execute(
+            "DELETE FROM documents WHERE source = ?1",
+            rusqlite::params![source],
+        )
+        .map_err(|e| OpenJarvisError::Io(std::io::Error::other(e.to_string())))?;
+
+        let mut doc_ids = Vec::with_capacity(documents.len());
+        for (content, metadata) in documents {
+            let doc_id = Uuid::new_v4().to_string();
+            let meta_str = metadata
+                .map(|m| serde_json::to_string(m).unwrap_or_default())
+                .unwrap_or_else(|| "{}".to_string());
+
+            tx.execute(
+                "INSERT INTO documents (id, content, source, metadata)
+                 VALUES (?1, ?2, ?3, ?4)",
+                rusqlite::params![doc_id, content, source, meta_str],
+            )
+            .map_err(|e| OpenJarvisError::Io(std::io::Error::other(e.to_string())))?;
+
+            let rowid = tx.last_insert_rowid();
+            tx.execute(
+                "INSERT INTO documents_fts (rowid, content, source) VALUES (?1, ?2, ?3)",
+                rusqlite::params![rowid, content, source],
+            )
+            .map_err(|e| OpenJarvisError::Io(std::io::Error::other(e.to_string())))?;
+            doc_ids.push(doc_id);
+        }
+
+        tx.commit()
+            .map_err(|e| OpenJarvisError::Io(std::io::Error::other(e.to_string())))?;
+        Ok(doc_ids)
+    }
 }
 
 impl MemoryBackend for SQLiteMemory {
@@ -306,6 +357,40 @@ mod tests {
         assert_eq!(mem.count().unwrap(), 0);
     }
 
+    #[test]
+    fn test_sqlite_replace_source_is_idempotent() {
+        let mem = SQLiteMemory::in_memory().unwrap();
+
+        mem.replace_source("notes.txt", &[("old project notes", None)])
+            .unwrap();
+        assert_eq!(mem.count().unwrap(), 1);
+
+        mem.replace_source("notes.txt", &[("updated project notes", None)])
+            .unwrap();
+        assert_eq!(mem.count().unwrap(), 1);
+
+        assert!(mem.retrieve("old", 5).unwrap().is_empty());
+        let updated = mem.retrieve("updated", 5).unwrap();
+        assert_eq!(updated.len(), 1);
+        assert_eq!(updated[0].source, "notes.txt");
+    }
+
+    #[test]
+    fn test_sqlite_replace_source_preserves_other_sources() {
+        let mem = SQLiteMemory::in_memory().unwrap();
+        mem.store("keep this manual", "manual.txt", None).unwrap();
+        mem.replace_source("notes.txt", &[("old project notes", None)])
+            .unwrap();
+
+        mem.replace_source("notes.txt", &[("updated project notes", None)])
+            .unwrap();
+
+        assert_eq!(mem.count().unwrap(), 2);
+        let manual = mem.retrieve("manual", 5).unwrap();
+        assert_eq!(manual.len(), 1);
+        assert_eq!(manual[0].source, "manual.txt");
+    }
+
     #[test]
     fn test_sqlite_case_insensitive_search() {
         let mem = SQLiteMemory::in_memory().unwrap();

src/openjarvis/cli/memory_cmd.py

@@ -89,15 +89,39 @@ def index(
 
     mem = _get_backend(backend)
     try:
-        for chunk in track(chunks, description="Storing chunks...", console=console):
-            mem.store(
-                chunk.content,
-                source=chunk.source,
-                metadata={
-                    "offset": chunk.offset,
-                    "index": chunk.index,
-                },
-            )
+        replace_source = getattr(mem, "replace_source", None)
+        if callable(replace_source):
+            documents_by_source = {}
+            for chunk in chunks:
+                documents_by_source.setdefault(chunk.source, []).append(
+                    (
+                        chunk.content,
+                        {
+                            "offset": chunk.offset,
+                            "index": chunk.index,
+                        },
+                    )
+                )
+            for source, documents in track(
+                documents_by_source.items(),
+                description="Replacing sources...",
+                console=console,
+            ):
+                replace_source(source, documents)
+        else:
+            for chunk in track(
+                chunks,
+                description="Storing chunks...",
+                console=console,
+            ):
+                mem.store(
+                    chunk.content,
+                    source=chunk.source,
+                    metadata={
+                        "offset": chunk.offset,
+                        "index": chunk.index,
+                    },
+                )
     finally:
         if hasattr(mem, "close"):
             mem.close()

src/openjarvis/tools/storage/sqlite.py

@@ -95,6 +95,29 @@ class SQLiteMemory(MemoryBackend):
         )
         return doc_id
 
+    def replace_source(
+        self,
+        source: str,
+        documents: List[tuple[str, Optional[Dict[str, Any]]]],
+    ) -> List[str]:
+        """Atomically replace all documents associated with *source*."""
+        payload = [
+            (content, json.dumps(metadata) if metadata else None)
+            for content, metadata in documents
+        ]
+        doc_ids = self._rust_impl.replace_source(source, payload)
+        bus = get_event_bus()
+        for doc_id in doc_ids:
+            bus.publish(
+                EventType.MEMORY_STORE,
+                {
+                    "backend": self.backend_id,
+                    "doc_id": doc_id,
+                    "source": source,
+                },
+            )
+        return doc_ids
+
     def retrieve(
         self,
         query: str,

tests/cli/test_memory_cmd.py

@@ -40,6 +40,35 @@ def test_memory_index_file(tmp_path: Path, monkeypatch):
     assert "Indexed" in result.output or "chunk" in result.output
 
 
+def test_memory_index_replaces_existing_source(tmp_path: Path, monkeypatch):
+    """Re-indexing a file replaces its previous chunks."""
+    _register_sqlite()
+    db_path = str(tmp_path / "mem.db")
+    doc = tmp_path / "doc.txt"
+    doc.write_text(" ".join(["legacy"] * 100), encoding="utf-8")
+
+    mod = importlib.import_module("openjarvis.cli.memory_cmd")
+    monkeypatch.setattr(
+        mod,
+        "_get_backend",
+        lambda b=None: SQLiteMemory(db_path=db_path),
+    )
+
+    first = CliRunner().invoke(cli, ["memory", "index", str(doc)])
+    assert first.exit_code == 0
+
+    doc.write_text(" ".join(["updated"] * 100), encoding="utf-8")
+    second = CliRunner().invoke(cli, ["memory", "index", str(doc)])
+    assert second.exit_code == 0
+
+    backend = SQLiteMemory(db_path=db_path)
+    assert backend.count() == 1
+    assert backend.retrieve("legacy") == []
+    updated = backend.retrieve("updated")
+    assert len(updated) == 1
+    assert updated[0].source == str(doc)
+
+
 def test_memory_index_nonexistent(tmp_path: Path):
     """Indexing a nonexistent path should fail."""
     _register_sqlite()