fix(kb): replace stale chunk rows on re-chunk (issue #199)

When users re-chunk a document with different parameters, the old chunk
rows (different config_hash) must be removed. Introduce
`replace_chunks()` in the storage abstraction layer that safely inserts
new rows first then deletes the old generation (crash-safe: worst case
is brief duplicate data, never data loss).

Also:
- Add scope-key whitelist validation to prevent filter injection
- Add concurrency note (last-write-wins) to chunk_document docstring
- Add 4 unit tests for replace_chunks edge cases
- Make separator-vs-default integration test deterministic (tmp_path)

Refs: #199

Author: sqhyz55

src/xagent/core/tools/core/RAG_tools/chunk/chunk_document.py

@@ -87,6 +87,13 @@ def chunk_document(
     """
     Chunk parsed paragraphs and write to chunks table.
 
+    Concurrency note:
+        This function uses **last-write-wins** semantics. If two concurrent
+        calls target the same (collection, doc_id, parse_hash) with different
+        parameters, the last one to reach ``_write_chunks_to_db`` will replace
+        the other's output. Callers requiring mutual exclusion should hold a
+        collection-level lock before invoking this function.
+
     Args:
         collection: Collection name for data isolation
         doc_id: Document ID whose parsed result to chunk
@@ -474,7 +481,12 @@ def _write_chunks_to_db(
     user_id: Optional[int] = None,
     is_admin: bool = False,
 ) -> bool:
-    """Write chunk records to database using abstraction layer."""
+    """Write chunk records to database using abstraction layer.
+
+    Replaces any existing rows for the same document parse (collection, doc_id,
+    parse_hash) and tenancy scope before inserting, so a new ``config_hash`` does
+    not leave stale chunks from a previous configuration (issue #199).
+    """
     try:
         rows = []
         for chunk in chunks:
@@ -503,9 +515,17 @@ def _write_chunks_to_db(
         if not rows:
             return False
 
-        # Use abstraction layer for upsert
         vector_store = get_vector_index_store()
-        vector_store.upsert_chunks(rows)
+        vector_store.replace_chunks(
+            rows,
+            replace_scope={
+                "collection": collection,
+                "doc_id": doc_id,
+                "parse_hash": parse_hash,
+            },
+            user_id=user_id,
+            is_admin=is_admin,
+        )
 
         logger.info(
             "Chunk records written to database: doc_id=%s, parse_hash=%s, config_hash=%s",

src/xagent/core/tools/core/RAG_tools/storage/contracts.py

@@ -713,6 +713,30 @@ def upsert_chunks(self, records: List[Dict[str, Any]]) -> None:
             records: List of chunk record dictionaries to upsert.
         """
 
+    @abstractmethod
+    def replace_chunks(
+        self,
+        records: List[Dict[str, Any]],
+        *,
+        replace_scope: Dict[str, Any],
+        user_id: Optional[int] = None,
+        is_admin: bool = False,
+    ) -> None:
+        """Replace chunk records within a scope (delete old + insert new).
+
+        Deletes all existing chunk rows matching *replace_scope* (and tenancy
+        filters), then inserts *records*. This guarantees that re-chunking with
+        different parameters does not leave stale rows from a previous
+        configuration (issue #199).
+
+        Args:
+            records: New chunk records to insert after deletion.
+            replace_scope: Filter dict (e.g. collection, doc_id, parse_hash)
+                identifying rows to delete before inserting.
+            user_id: Optional user ID for multi-tenancy scoped deletion.
+            is_admin: Whether the caller can operate across tenants.
+        """
+
     @abstractmethod
     def upsert_embeddings(self, model_tag: str, records: List[Dict[str, Any]]) -> None:
         """Upsert embedding records (sync).

src/xagent/core/tools/core/RAG_tools/storage/lancedb_stores.py

@@ -1448,6 +1448,68 @@ def upsert_chunks(self, records: List[Dict[str, Any]]) -> None:
             _safe_close_table(table)
         self.invalidate_table_cache("chunks")
 
+    _REPLACE_CHUNKS_ALLOWED_KEYS: frozenset = frozenset(
+        {"collection", "doc_id", "parse_hash"}
+    )
+
+    def replace_chunks(
+        self,
+        records: List[Dict[str, Any]],
+        *,
+        replace_scope: Dict[str, Any],
+        user_id: Optional[int] = None,
+        is_admin: bool = False,
+    ) -> None:
+        """Replace chunk records within a scope (delete old + insert new).
+
+        Safety: inserts new records first, then deletes rows that do NOT belong
+        to the new generation. This avoids data loss if the process crashes
+        between the two operations (worst case: brief duplicate data, not zero
+        data).
+        """
+        from ..LanceDB.schema_manager import _safe_close_table, ensure_chunks_table
+
+        for key in replace_scope:
+            if key not in self._REPLACE_CHUNKS_ALLOWED_KEYS:
+                raise ValueError(f"Invalid replace_scope column: {key}")
+
+        conn = self._get_connection()
+        ensure_chunks_table(conn)
+        table = conn.open_table("chunks")
+        try:
+            # Step 1: Insert new records (safe — idempotent add)
+            if records:
+                table.add(records)
+
+            # Step 2: Build delete filter targeting old generations
+            scope_parts = [
+                f"{k} == '{escape_lancedb_string(str(v))}'"
+                for k, v in replace_scope.items()
+            ]
+            base_filter = " AND ".join(scope_parts)
+
+            user_filter = UserPermissions.get_user_filter(user_id, is_admin)
+            if base_filter and user_filter:
+                delete_expr = f"({base_filter}) AND ({user_filter})"
+            elif user_filter:
+                delete_expr = user_filter
+            else:
+                delete_expr = base_filter
+
+            # Exclude newly inserted chunk_ids from deletion
+            if records and delete_expr:
+                new_ids = [r["chunk_id"] for r in records]
+                id_list = ", ".join(
+                    f"'{escape_lancedb_string(cid)}'" for cid in new_ids
+                )
+                delete_expr = f"({delete_expr}) AND chunk_id NOT IN ({id_list})"
+
+            if delete_expr:
+                table.delete(delete_expr)
+        finally:
+            _safe_close_table(table)
+        self.invalidate_table_cache("chunks")
+
     def upsert_embeddings(self, model_tag: str, records: List[Dict[str, Any]]) -> None:
         """Upsert embedding records to LanceDB with fallback pattern.
 

tests/core/tools/core/RAG_tools/chunk/test_chunk_document.py

@@ -810,7 +810,7 @@ def test_chunk_config_hash_idempotency(
     def test_chunk_separators_create_new_version(
         self, temp_lancedb_dir, test_collection, test_doc_id
     ):
-        """Test that different separators create new version."""
+        """Re-chunking with different separators replaces prior rows (single config_hash)."""
         # Step 1: Register and parse document
         txt_path = "tests/resources/test_files/test.txt"
         register_document(
@@ -852,11 +852,9 @@ def test_chunk_separators_create_new_version(
             user_id=1,
         )
 
-        # Both should write (different versions)
         assert chunk_result1["created"] is True
         assert chunk_result2["created"] is True
 
-        # Verify database has two different config_hash versions
         from xagent.core.tools.core.RAG_tools.storage.factory import (
             get_vector_store_raw_connection,
         )
@@ -871,14 +869,11 @@ def test_chunk_separators_create_new_version(
             .to_pandas()
         )
 
-        # Should have two different config_hash values
+        # Only the latest chunking generation is kept for this parse (issue #199).
         config_hashes = df["config_hash"].unique()
-        assert len(config_hashes) == 2
-
-        # Each version should have > 0 rows
-        for config_hash in config_hashes:
-            version_rows = df[df["config_hash"] == config_hash]
-            assert len(version_rows) > 0
+        assert len(config_hashes) == 1
+        assert len(df) == chunk_result2["chunk_count"]
+        assert len(df) > 0
 
     def test_chunk_recursive_custom_separators_integration(
         self, temp_lancedb_dir, test_collection, test_doc_id
@@ -934,13 +929,19 @@ def test_chunk_recursive_custom_separators_integration(
         )
 
     def test_chunk_recursive_custom_separators_vs_default_different_result(
-        self, temp_lancedb_dir, test_collection, test_doc_id
+        self, temp_lancedb_dir, test_collection, test_doc_id, tmp_path
     ):
-        """Integration: custom separators produce different chunk count or config than default."""
-        txt_path = "tests/resources/test_files/test.txt"
+        """Integration: default then custom separators; DB keeps only the last generation."""
+        # Generate deterministic long text to guarantee different chunk counts
+        lines = [
+            f"Line {i}: The quick brown fox jumps over the lazy dog." for i in range(20)
+        ]
+        long_doc = tmp_path / "long_doc.txt"
+        long_doc.write_text("\n".join(lines), encoding="utf-8")
+
         register_document(
             collection=test_collection,
-            source_path=txt_path,
+            source_path=str(long_doc),
             doc_id=test_doc_id,
             user_id=1,
         )
@@ -953,13 +954,14 @@ def test_chunk_recursive_custom_separators_vs_default_different_result(
         )
         parse_hash = parse_result["parse_hash"]
 
+        # Large chunk_size → few chunks; small chunk_size with newline separator → many chunks.
         chunk_default = chunk_document(
             collection=test_collection,
             doc_id=test_doc_id,
             parse_hash=parse_hash,
             chunk_strategy=ChunkStrategy.RECURSIVE,
-            chunk_size=50,
-            chunk_overlap=10,
+            chunk_size=500,
+            chunk_overlap=0,
             separators=None,
             user_id=1,
         )
@@ -968,14 +970,16 @@ def test_chunk_recursive_custom_separators_vs_default_different_result(
             doc_id=test_doc_id,
             parse_hash=parse_hash,
             chunk_strategy=ChunkStrategy.RECURSIVE,
-            chunk_size=50,
-            chunk_overlap=10,
-            separators=["。", "\n"],
+            chunk_size=25,
+            chunk_overlap=5,
+            separators=["\n"],
             user_id=1,
         )
         assert chunk_default["created"] is True
         assert chunk_custom["created"] is True
-        # Different separators must yield different config_hash (hence different version)
+        assert chunk_default["chunk_count"] != chunk_custom["chunk_count"], (
+            "Expected distinct chunking outcomes so the replacement semantics are observable"
+        )
         from xagent.core.tools.core.RAG_tools.storage.factory import (
             get_vector_store_raw_connection,
         )
@@ -990,9 +994,8 @@ def test_chunk_recursive_custom_separators_vs_default_different_result(
             .to_pandas()
         )
         config_hashes = df["config_hash"].unique()
-        assert len(config_hashes) == 2, (
-            "Default and custom separators should produce two distinct chunk versions"
-        )
+        assert len(config_hashes) == 1
+        assert len(df) == chunk_custom["chunk_count"]
 
     def test_chunk_row_level_hash_uniqueness(
         self, temp_lancedb_dir, test_collection, test_doc_id