fix(kb): scope collection rename by tenant and block shared-name conflicts

Reject non-admin renames when multiple users share a collection name, require
admin target_user_id for scoped renames, and align metadata/config updates with
per-user vector renames to prevent cross-tenant control-plane corruption.

Author: sqhyz55

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

@@ -334,16 +334,36 @@ async def delete_collection_metadata(
         """Delete persisted metadata/config rows for a collection."""
 
     @abstractmethod
-    async def rename_collection(self, old_name: str, new_name: str) -> None:
-        """Rename persisted control-plane keys after a data-plane collection rename.
+    async def count_users_with_collection_config(self, collection_name: str) -> int:
+        """Count distinct users with a ``collection_config`` row for ``collection_name``.
 
-        Updates rows that gate :meth:`list_collections` visibility (for example
-        per-tenant config rows and aggregate metadata) so they stay aligned with
-        vector tables when the ``collection`` / ``name`` fields change.
+        Args:
+            collection_name: Collection name (sanitized by the caller).
+
+        Returns:
+            Number of distinct ``user_id`` values occupying the name.
+        """
+
+    @abstractmethod
+    async def rename_collection(
+        self,
+        old_name: str,
+        new_name: str,
+        *,
+        user_id: int,
+        is_admin: bool = False,
+    ) -> None:
+        """Rename persisted control-plane keys for one user's collection scope.
+
+        Updates the caller's ``collection_config`` row. Global ``collection_metadata``
+        is renamed only when this user is the sole config occupant of ``old_name``.
 
         Args:
             old_name: Previous collection name (sanitized by the caller).
             new_name: Target collection name (sanitized by the caller).
+            user_id: User whose config (and optionally metadata) should be renamed.
+            is_admin: Whether the operation is performed by an admin on behalf of
+                ``user_id`` (does not broaden vector/config scope beyond ``user_id``).
         """
 
     @abstractmethod

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

@@ -80,42 +80,94 @@ async def delete_collection(self, collection_name: str) -> None:
         except Exception as exc:
             logger.debug("Failed to delete collection metadata: %s", exc)
 
-    async def rename_collection(self, old_name: str, new_name: str) -> None:
-        """Rename ``collection_config`` and ``collection_metadata`` keys.
+    async def count_users_with_collection_config(self, collection_name: str) -> int:
+        """Count distinct ``user_id`` rows in ``collection_config`` for a collection name."""
+        from ..LanceDB.schema_manager import (
+            _safe_close_table,
+            ensure_collection_config_table,
+        )
+
+        conn = await self._get_connection()
+        ensure_collection_config_table(conn)
+
+        safe_collection = escape_lancedb_string(collection_name)
+        config_table = None
+        try:
+            config_table = conn.open_table("collection_config")
+            rows = (
+                config_table.search()
+                .where(f"collection = '{safe_collection}'")
+                .to_arrow()
+                .to_pylist()
+            )
+        except Exception as exc:  # noqa: BLE001
+            logger.debug(
+                "Failed to count collection config occupants for %s: %s",
+                collection_name,
+                exc,
+            )
+            return 0
+        finally:
+            _safe_close_table(config_table)
+
+        occupant_ids = {
+            int(row["user_id"]) for row in rows if row.get("user_id") is not None
+        }
+        return len(occupant_ids)
+
+    async def rename_collection(
+        self,
+        old_name: str,
+        new_name: str,
+        *,
+        user_id: int,
+        is_admin: bool = False,
+    ) -> None:
+        """Rename control-plane keys for a single user's collection scope.
 
         See :meth:`MetadataStore.rename_collection`.
         """
+        from ..core.schemas import CollectionInfo
         from ..LanceDB.schema_manager import (
             _safe_close_table,
             ensure_collection_config_table,
         )
 
+        del (
+            is_admin
+        )  # Scope is always limited to ``user_id``; flag is for callers only.
+
         conn = await self._get_connection()
         await self.ensure_collection_metadata_table()
         ensure_collection_config_table(conn)
 
         safe_old = escape_lancedb_string(old_name)
         now = datetime.now(timezone.utc).replace(tzinfo=None)
+        occupant_count = await self.count_users_with_collection_config(old_name)
 
         config_table = None
         try:
             config_table = conn.open_table("collection_config")
             config_table.update(
-                f"collection = '{safe_old}'",
+                f"collection = '{safe_old}' AND user_id = {int(user_id)}",
                 {"collection": new_name, "updated_at": now},
             )
         finally:
             _safe_close_table(config_table)
 
-        meta_table = None
-        try:
-            meta_table = conn.open_table("collection_metadata")
-            meta_table.update(
-                f"name = '{safe_old}'",
-                {"name": new_name, "updated_at": now},
-            )
-        finally:
-            _safe_close_table(meta_table)
+        if occupant_count <= 1:
+            meta_table = None
+            try:
+                meta_table = conn.open_table("collection_metadata")
+                meta_table.update(
+                    f"name = '{safe_old}'",
+                    {"name": new_name, "updated_at": now},
+                )
+            finally:
+                _safe_close_table(meta_table)
+            return
+
+        await self.save_collection(CollectionInfo(name=new_name))
 
     async def save_collection(self, collection: CollectionInfo) -> None:
         from ..LanceDB.schema_manager import _safe_close_table

src/xagent/web/api/kb.py

@@ -77,7 +77,10 @@
 )
 from ...core.tools.core.RAG_tools.progress import get_progress_manager
 from ...core.tools.core.RAG_tools.storage.contracts import DocumentRecord
-from ...core.tools.core.RAG_tools.storage.factory import get_vector_index_store
+from ...core.tools.core.RAG_tools.storage.factory import (
+    get_metadata_store,
+    get_vector_index_store,
+)
 from ...core.tools.core.RAG_tools.utils.string_utils import (
     generate_deterministic_doc_id,
 )
@@ -3916,6 +3919,10 @@ def _resolve_list_documents_match() -> Optional[ResolvedDocumentMatch]:
 async def rename_collection_api(
     collection_name: str,
     new_name: str = Form(..., description="New collection name"),
+    target_user_id: Optional[int] = Form(
+        None,
+        description="Admin only: user whose collection scope to rename",
+    ),
     _user: User = Depends(get_current_user),
     db: Session = Depends(get_db),
 ) -> dict:
@@ -3924,6 +3931,7 @@ async def rename_collection_api(
     Args:
         collection_name: Current collection name
         new_name: New collection name
+        target_user_id: Required for admins; scopes rename to that user's data only
 
     Returns:
         Success message
@@ -3933,13 +3941,20 @@ async def rename_collection_api(
         load_ingestion_status,
         write_ingestion_status,
     )
-    from ...core.tools.core.RAG_tools.storage.factory import (
-        get_metadata_store,
-        get_vector_index_store,
-    )
 
+    metadata_store = get_metadata_store()
     vector_store = get_vector_index_store()
 
+    if bool(_user.is_admin):
+        if target_user_id is None:
+            raise HTTPException(
+                status_code=422,
+                detail="Admin rename requires target_user_id form field",
+            )
+        scope_user_id = int(target_user_id)
+    else:
+        scope_user_id = int(_user.id)
+
     if not new_name or not new_name.strip():
         raise HTTPException(
             status_code=422,
@@ -3961,12 +3976,40 @@ async def rename_collection_api(
     if safe_new_collection == safe_old_collection:
         return {"status": "success", "message": "Collection name unchanged"}
 
+    is_admin_rename = bool(_user.is_admin)
+    if not is_admin_rename:
+        occupant_count = await metadata_store.count_users_with_collection_config(
+            safe_old_collection
+        )
+        if occupant_count > 1:
+            raise HTTPException(
+                status_code=409,
+                detail=(
+                    "Collection name is shared by multiple users; "
+                    "rename is not allowed. Contact an administrator."
+                ),
+            )
+
     # Access control check
-    await _ensure_collection_access(safe_old_collection, _user, hide_missing=False)
+    if is_admin_rename:
+        visible_for_scope = await _list_collections_with_retry(
+            user_id=scope_user_id,
+            is_admin=False,
+            stage="rename_list_visible_collections_for_target_user",
+        )
+        if not any(
+            c.name == safe_old_collection for c in visible_for_scope.collections
+        ):
+            raise HTTPException(
+                status_code=404,
+                detail=f"Collection not found for user: {safe_old_collection}",
+            )
+    else:
+        await _ensure_collection_access(safe_old_collection, _user, hide_missing=False)
 
     # Validate that target collection doesn't exist or user has access
     visible_for_user = await _list_collections_with_retry(
-        user_id=int(_user.id),
+        user_id=scope_user_id,
         is_admin=False,
         stage="rename_list_visible_collections",
     )
@@ -3993,8 +4036,8 @@ async def rename_collection_api(
     new_collection_dir: Optional[Path] = None
     collection_records = vector_store.list_document_records(
         collection_name=safe_old_collection,
-        user_id=int(_user.id),
-        is_admin=bool(_user.is_admin),
+        user_id=scope_user_id,
+        is_admin=False,
     )
     collection_file_ids = {
         file_id
@@ -4006,7 +4049,7 @@ async def rename_collection_api(
 
     physical_rename = rename_collection_storage(
         db,
-        user_id=int(_user.id),
+        user_id=scope_user_id,
         old_collection_name=safe_old_collection,
         new_collection_name=safe_new_collection,
         collection_file_ids=collection_file_ids,
@@ -4031,30 +4074,33 @@ async def rename_collection_api(
         )
 
     # Step 2: Update collection name in all tables (documents, parses, chunks, embeddings)
-    # Use storage abstraction layer which handles all tables including embeddings
-    vector_store = get_vector_index_store()
     warnings.extend(
         vector_store.rename_collection_data(
             collection_name=safe_old_collection,
             new_name=safe_new_collection,
-            user_id=int(_user.id),
-            is_admin=bool(_user.is_admin),
+            user_id=scope_user_id,
+            is_admin=False,
         )
     )
 
     try:
-        metadata_store = get_metadata_store()
         await metadata_store.rename_collection(
             old_name=safe_old_collection,
             new_name=safe_new_collection,
+            user_id=scope_user_id,
+            is_admin=is_admin_rename,
         )
     except Exception as e:
         logger.warning("Failed to rename metadata store keys: %s", e)
         warnings.append(f"Failed to rename collection metadata: {e}")
 
     # Migrate ingestion status from old collection name to new
     try:
-        status_entries = load_ingestion_status(collection=safe_old_collection)
+        status_entries = load_ingestion_status(
+            collection=safe_old_collection,
+            user_id=scope_user_id,
+            is_admin=False,
+        )
         for entry in status_entries:
             doc_id = entry.get("doc_id")
             if doc_id:
@@ -4064,8 +4110,14 @@ async def rename_collection_api(
                     status=entry.get("status", "pending"),
                     message=entry.get("message", ""),
                     parse_hash=entry.get("parse_hash", ""),
+                    user_id=scope_user_id,
+                )
+                clear_ingestion_status(
+                    safe_old_collection,
+                    doc_id,
+                    user_id=scope_user_id,
+                    is_admin=False,
                 )
-                clear_ingestion_status(safe_old_collection, doc_id)
     except Exception as e:
         logger.warning("Failed to update ingestion status: %s", e)
         warnings.append(f"Failed to update ingestion status: {e}")

tests/core/tools/core/RAG_tools/management/test_collections.py

@@ -1067,3 +1067,57 @@ def test_delete_collection_removes_metadata_table_entry(temp_lancedb_dir: str) -
     after_table = conn.open_table("collection_metadata")
     after = after_table.search().where(f"name = '{collection}'").to_list()
     assert after == []
+
+
+@pytest.mark.asyncio
+async def test_metadata_rename_collection_preserves_other_users_config_when_shared(
+    temp_lancedb_dir: str,
+) -> None:
+    """Tenant-scoped metadata rename must not retarget another user's config row."""
+    from src.xagent.core.tools.core.RAG_tools.core.schemas import CollectionInfo
+
+    collection = "shared_rename"
+    store = get_metadata_store()
+    await store.save_collection(CollectionInfo(name=collection))
+    await store.save_collection_config(collection, "{}", user_id=1)
+    await store.save_collection_config(collection, "{}", user_id=2)
+
+    await store.rename_collection(
+        collection,
+        "user1_new",
+        user_id=1,
+        is_admin=False,
+    )
+
+    assert await store.get_collection_config("user1_new", user_id=1) == "{}"
+    assert await store.get_collection_config(collection, user_id=2) == "{}"
+    assert await store.get_collection_config("user1_new", user_id=2) is None
+
+    listed = await list_collections(user_id=None, is_admin=True)
+    names = {info.name for info in listed.collections}
+    assert collection in names
+    assert "user1_new" in names
+
+
+@pytest.mark.asyncio
+async def test_metadata_rename_collection_admin_scopes_to_target_user_only(
+    temp_lancedb_dir: str,
+) -> None:
+    """Admin rename for one user must leave other occupants on the old name."""
+    from src.xagent.core.tools.core.RAG_tools.core.schemas import CollectionInfo
+
+    collection = "shared_admin_rename"
+    store = get_metadata_store()
+    await store.save_collection(CollectionInfo(name=collection))
+    await store.save_collection_config(collection, "{}", user_id=10)
+    await store.save_collection_config(collection, "{}", user_id=20)
+
+    await store.rename_collection(
+        collection,
+        "user10_new",
+        user_id=10,
+        is_admin=True,
+    )
+
+    assert await store.get_collection_config("user10_new", user_id=10) == "{}"
+    assert await store.get_collection_config(collection, user_id=20) == "{}"