Merge pull request '🐾 fix(api): workspace DELETE must not report success it didn't perform' (#11) from fix/workspace-delete-orphan-libraries into main

Reviewed-on: #11

mnemosyne/library/api/views.py

@@ -10,6 +10,7 @@ import os
 
 from django.core.files.base import ContentFile
 from django.core.files.storage import default_storage
+from neomodel.exceptions import UniqueProperty
 from rest_framework import status
 from rest_framework.decorators import api_view, parser_classes, permission_classes
 from rest_framework.parsers import FormParser, JSONParser, MultiPartParser
@@ -17,6 +18,11 @@ from rest_framework.permissions import IsAuthenticated
 from rest_framework.response import Response
 
 from library.content_types import get_library_type_config
+from library.services.library_delete import (
+    delete_collection_cascade,
+    delete_item_cascade,
+    delete_library_cascade,
+)
 from mcp_server.drf_auth import request_token_label
 
 from .serializers import (
@@ -51,7 +57,7 @@ def library_list_create(request):
     a per-library ``item_count``. Off by default because the count is a
     Cypher aggregate; on for the Daedalus-side registry poll.
     """
-    from library.models import Library
+    from library.models import Library, find_library_by_name_ci
 
     if request.method == "GET":
         include_workspace = request.GET.get("include_workspace", "true").lower() != "false"
@@ -85,13 +91,13 @@ def library_list_create(request):
     serializer.is_valid(raise_exception=True)
     data = serializer.validated_data
 
-    # Library.name is globally unique; reject collisions with a clean 409
-    # (uid + managed_by included so the caller can say who owns the name)
-    # instead of letting the unique-index save raise a 500.
-    try:
-        existing = Library.nodes.get(name=data["name"])
-    except Library.DoesNotExist:
-        existing = None
+    # Library names are unique. The Neo4j index is case-sensitive, but
+    # clients (Spelunker, humans) treat names case-insensitively, so the
+    # create-time check is case-insensitive too — otherwise "amazon connect"
+    # silently creates a near-duplicate of "Amazon Connect". Reject with a
+    # clean 409 (uid + managed_by included so the caller can say who owns
+    # the name) instead of letting the unique-index save raise a 500.
+    existing = find_library_by_name_ci(data["name"])
     if existing is not None:
         logger.warning(
             "library_create name_conflict name=%s existing_uid=%s caller=%s",
@@ -99,7 +105,7 @@ def library_list_create(request):
         )
         return Response(
             {
-                "detail": f"A library named '{data['name']}' already exists.",
+                "detail": f"A library named '{existing.name}' already exists.",
                 "code": "name_conflict",
                 "uid": existing.uid,
                 "managed_by": existing.managed_by_display or None,
@@ -127,7 +133,22 @@ def library_list_create(request):
             data.get("llm_context_prompt") or defaults["llm_context_prompt"]
         ),
     )
+    try:
         lib.save()
+    except UniqueProperty:
+        # Race between the pre-check and save — an exact-name twin landed
+        # in between. Same 409 shape, minus the loser's uid/manager.
+        logger.warning(
+            "library_create name_conflict (save race) name=%s caller=%s",
+            data["name"], request.user.username,
+        )
+        return Response(
+            {
+                "detail": f"A library named '{data['name']}' already exists.",
+                "code": "name_conflict",
+            },
+            status=status.HTTP_409_CONFLICT,
+        )
     return Response(LibrarySerializer(lib).data, status=status.HTTP_201_CREATED)
 
 
@@ -165,8 +186,16 @@ def library_detail(request, uid):
         lib.save()
         return Response(LibrarySerializer(lib).data)
 
-    # DELETE
-    lib.delete()
+    # DELETE — use the shared cascade so child nodes (Collections/Items/
+    # Chunks/Images) and orphan Concepts are removed too; a bare
+    # lib.delete() would leak them all.
+    result = delete_library_cascade(lib)
+    logger.info(
+        "Library deleted via API library_uid=%s name=%s items=%d "
+        "orphans_deleted=%d caller=%s",
+        result["library_uid"], result["name"], result["item_count"],
+        result["orphans_deleted"], request.user.username,
+    )
     return Response(status=status.HTTP_204_NO_CONTENT)
 
 
@@ -248,7 +277,13 @@ def collection_detail(request, uid):
         col.save()
         return Response(CollectionSerializer(col).data)
 
-    col.delete()
+    # DELETE — cascade Items/Chunks/Images too; a bare col.delete() leaks them.
+    result = delete_collection_cascade(col)
+    logger.info(
+        "Collection deleted via API collection_uid=%s name=%s items=%d caller=%s",
+        result["collection_uid"], result["name"], result["item_count"],
+        request.user.username,
+    )
     return Response(status=status.HTTP_204_NO_CONTENT)
 
 
@@ -329,7 +364,13 @@ def item_detail(request, uid):
         item.save()
         return Response(ItemSerializer(item).data)
 
-    item.delete()
+    # DELETE — cascade Chunks/Images/embeddings too; a bare item.delete()
+    # leaks them.
+    delete_item_cascade(item.uid)
+    logger.info(
+        "Item deleted via API item_uid=%s caller=%s",
+        item.uid, request.user.username,
+    )
     return Response(status=status.HTTP_204_NO_CONTENT)
 
 

mnemosyne/library/api/workspaces.py

@@ -67,7 +67,7 @@ def workspace_create(request):
     workspace (200) — not an error. The library_type is frozen at first
     create; subsequent calls are not allowed to change it.
     """
-    from library.models import Library
+    from library.models import Library, find_library_by_name_ci
 
     serializer = WorkspaceCreateSerializer(data=request.data)
     serializer.is_valid(raise_exception=True)
@@ -127,6 +127,28 @@ def workspace_create(request):
             status=status.HTTP_200_OK,
         )
 
+    # New workspace: reject a name already taken by any other library,
+    # case-insensitively — the Neo4j index is case-sensitive, so without
+    # this check "amazon connect" would silently coexist with an existing
+    # "Amazon Connect" and confuse every name-matching client.
+    name_taken = find_library_by_name_ci(data["name"])
+    if name_taken is not None:
+        logger.warning(
+            "workspace_create name_conflict workspace_id=%s name=%s "
+            "existing_uid=%s",
+            data["workspace_id"], data["name"], name_taken.uid,
+        )
+        return Response(
+            {
+                "detail": (
+                    f"A library named '{name_taken.name}' already exists in "
+                    "Mnemosyne."
+                ),
+                "code": "name_conflict",
+            },
+            status=status.HTTP_409_CONFLICT,
+        )
+
     defaults = get_library_type_config(data["library_type"])
     lib = Library(
         name=data["name"],
@@ -191,23 +213,40 @@ def workspace_detail_or_delete(request, workspace_id):
     except Library.DoesNotExist:
         lib = None
 
-    # Cross-user reads/writes look like "not found" — don't disclose
-    # existence across users.
-    if lib is not None and lib.owner_username != request.user.username:
-        lib = None
+    # Cross-user reads look like "not found" — don't disclose existence
+    # across users. DELETE is handled separately below: telling the caller
+    # "deleted" about a library we did not touch is what silently orphans
+    # libraries, and an orphan holds its globally-unique name forever.
+    unowned = lib is not None and lib.owner_username != request.user.username
 
     if request.method == "GET":
-        if lib is None:
+        if lib is None or unowned:
             return Response(
                 {"detail": "Workspace not found."},
                 status=status.HTTP_404_NOT_FOUND,
             )
         return Response(WorkspaceStatusSerializer(_serialize_workspace(lib)).data)
 
-    # DELETE — idempotent: a missing (or unowned) workspace returns 204.
+    # DELETE — idempotent only where nothing exists to delete.
     if lib is None:
         return Response(status=status.HTTP_204_NO_CONTENT)
 
+    if unowned:
+        # Still opaque about ownership, but never a false success: the
+        # caller must not record this workspace as cleaned up.
+        logger.warning(
+            "workspace_delete owner_conflict workspace_id=%s library_uid=%s "
+            "caller=%s",
+            workspace_id, lib.uid, request.user.username,
+        )
+        return Response(
+            {
+                "detail": "Workspace id is already in use.",
+                "code": "owner_conflict",
+            },
+            status=status.HTTP_409_CONFLICT,
+        )
+
     # Delete the Library and everything reachable + unique to it, plus
     # orphan-Concept GC. Shared with the admin/HTML delete path.
     result = delete_library_cascade(lib)

mnemosyne/library/models.py

@@ -63,6 +63,23 @@ def infer_legacy_manager(workspace_id):
     return "Kairos" if workspace_id.startswith("kairos-mail-") else "Daedalus"
 
 
+def find_library_by_name_ci(name):
+    """Case-insensitively find a Library by name, or None.
+
+    Parameterised Cypher rather than neomodel's ``iexact``, which embeds
+    the value in a regex and so breaks on names containing regex
+    metacharacters (e.g. "C++ Notes").
+    """
+    from neomodel import db
+
+    rows, _ = db.cypher_query(
+        "MATCH (l:Library) WHERE toLower(l.name) = toLower($name) "
+        "RETURN l LIMIT 1",
+        {"name": name},
+    )
+    return Library.inflate(rows[0][0]) if rows else None
+
+
 class Library(StructuredNode):
     """
     Top-level container representing a content library.

mnemosyne/library/services/library_delete.py

@@ -106,3 +106,99 @@ def delete_library_cascade(lib) -> dict:
         "item_s3_keys": item_s3_keys,
         "orphans_deleted": orphans_deleted,
     }
+
+
+def delete_collection_cascade(col) -> dict:
+    """Delete ``col`` and all content reachable and unique to it.
+
+    Removes the Collection's Items with their Chunks, Images, and
+    ImageEmbeddings, then the Collection itself. No orphan-Concept GC —
+    that invariant belongs to library-level deletes only (see
+    :func:`delete_library_cascade`); Concepts orphaned here are collected
+    on the next library delete.
+
+    :param col: A ``library.models.Collection`` node instance.
+    :returns: Dict with ``collection_uid``, ``name``, ``item_count``, and
+        ``item_s3_keys`` (list of ``(uid, s3_key)`` for async S3 cleanup).
+    """
+    collection_uid = col.uid
+    collection_name = col.name
+
+    s3_rows, _ = db.cypher_query(
+        "MATCH (col:Collection {uid: $uid})-[:CONTAINS]->(i:Item) "
+        "RETURN i.uid, i.s3_key",
+        {"uid": collection_uid},
+    )
+    item_s3_keys = [(r[0], r[1]) for r in s3_rows if r[1]]
+
+    db.cypher_query(
+        """
+        MATCH (col:Collection {uid: $uid})-[:CONTAINS]->(i:Item)
+        -[:HAS_CHUNK]->(c:Chunk)
+        DETACH DELETE c
+        """,
+        {"uid": collection_uid},
+    )
+    db.cypher_query(
+        """
+        MATCH (col:Collection {uid: $uid})-[:CONTAINS]->(i:Item)
+        -[:HAS_IMAGE]->(img:Image)
+        OPTIONAL MATCH (img)-[:HAS_EMBEDDING]->(emb:ImageEmbedding)
+        DETACH DELETE img, emb
+        """,
+        {"uid": collection_uid},
+    )
+    db.cypher_query(
+        """
+        MATCH (col:Collection {uid: $uid})-[:CONTAINS]->(i:Item)
+        DETACH DELETE i
+        """,
+        {"uid": collection_uid},
+    )
+    db.cypher_query(
+        "MATCH (col:Collection {uid: $uid}) DETACH DELETE col",
+        {"uid": collection_uid},
+    )
+
+    logger.info(
+        "Collection cascade-deleted collection_uid=%s name=%s items=%d",
+        collection_uid, collection_name, len(item_s3_keys),
+    )
+
+    return {
+        "collection_uid": collection_uid,
+        "name": collection_name,
+        "item_count": len(item_s3_keys),
+        "item_s3_keys": item_s3_keys,
+    }
+
+
+def delete_item_cascade(item_uid: str) -> dict:
+    """Delete the Item ``item_uid`` with its Chunks, Images, and embeddings.
+
+    Keyed on the uid (not a node instance) so the ingest supersede path in
+    ``library.tasks`` can share it. No orphan-Concept GC — see
+    :func:`delete_collection_cascade`.
+
+    :returns: Dict with ``item_uid`` and ``s3_key`` (empty string when the
+        item had no stored file) for async S3 cleanup.
+    """
+    s3_rows, _ = db.cypher_query(
+        "MATCH (i:Item {uid: $uid}) RETURN i.s3_key",
+        {"uid": item_uid},
+    )
+    s3_key = (s3_rows[0][0] or "") if s3_rows else ""
+
+    db.cypher_query(
+        """
+        MATCH (i:Item {uid: $uid})
+        OPTIONAL MATCH (i)-[:HAS_CHUNK]->(c:Chunk)
+        OPTIONAL MATCH (i)-[:HAS_IMAGE]->(img:Image)
+        OPTIONAL MATCH (img)-[:HAS_EMBEDDING]->(emb:ImageEmbedding)
+        DETACH DELETE c, img, emb, i
+        """,
+        {"uid": item_uid},
+    )
+
+    logger.info("Item cascade-deleted item_uid=%s", item_uid)
+    return {"item_uid": item_uid, "s3_key": s3_key}

mnemosyne/library/tasks.py

@@ -503,16 +503,9 @@ def ingest_from_daedalus(self, job_id: str):
 
 def _delete_item_and_chunks(item_uid: str):
     """Delete an Item, its chunks, and its images. Concept GC is workspace-delete only."""
-    db.cypher_query(
-        """
-        MATCH (i:Item {uid: $uid})
-        OPTIONAL MATCH (i)-[:HAS_CHUNK]->(c:Chunk)
-        OPTIONAL MATCH (i)-[:HAS_IMAGE]->(img:Image)
-        OPTIONAL MATCH (img)-[:HAS_EMBEDDING]->(emb:ImageEmbedding)
-        DETACH DELETE c, img, emb, i
-        """,
-        {"uid": item_uid},
-    )
+    from library.services.library_delete import delete_item_cascade
+
+    delete_item_cascade(item_uid)
 
 
 def _resolve_or_create_default_collection(lib, collection_uid: str = ""):

mnemosyne/library/tests/test_library_api.py

@@ -0,0 +1,148 @@
+"""Tests for the plain library REST endpoints beyond create.
+
+Currently covers the DELETE cascades: library, collection, and item DELETE
+endpoints must go through the shared ``library_delete`` service functions —
+bare ``.delete()`` calls leak child nodes (Collections, Items, Chunks,
+Images) and, for libraries, skip orphan-Concept GC. Neo4j is stubbed via
+``sys.modules``, same style as ``test_managed_by.py``.
+"""
+
+from __future__ import annotations
+
+from types import SimpleNamespace
+from unittest.mock import MagicMock, patch
+
+from django.contrib.auth import get_user_model
+from django.test import TestCase
+from rest_framework.test import APIClient
+
+User = get_user_model()
+
+
+def _fake_node_cls(instance):
+    """A neomodel-class stand-in whose ``nodes.get`` returns ``instance``.
+
+    ``instance=None`` makes ``nodes.get`` raise the class's DoesNotExist.
+    """
+    fake_nodes = MagicMock()
+
+    class DoesNotExist(Exception):
+        pass
+
+    if instance is None:
+        fake_nodes.get.side_effect = DoesNotExist()
+    else:
+        fake_nodes.get.return_value = instance
+    return SimpleNamespace(nodes=fake_nodes, DoesNotExist=DoesNotExist)
+
+
+class LibraryApiDeleteTests(TestCase):
+    """DELETE on the plain library endpoint cascades."""
+
+    def setUp(self):
+        self.user = User.objects.create_user(username="op", password="pw")
+        self.client = APIClient()
+        self.client.force_authenticate(user=self.user)
+
+    def _fake_models_module(self, lib):
+        return SimpleNamespace(Library=_fake_node_cls(lib))
+
+    def test_delete_uses_shared_cascade(self):
+        lib = SimpleNamespace(uid="lib-1", name="Docs")
+        cascade_result = {
+            "library_uid": "lib-1",
+            "name": "Docs",
+            "item_count": 3,
+            "item_s3_keys": [],
+            "orphans_deleted": 1,
+        }
+        with patch.dict(
+            "sys.modules", {"library.models": self._fake_models_module(lib)}
+        ), patch(
+            "library.api.views.delete_library_cascade",
+            return_value=cascade_result,
+        ) as mock_cascade:
+            response = self.client.delete("/library/api/libraries/lib-1/")
+
+        self.assertEqual(response.status_code, 204)
+        mock_cascade.assert_called_once_with(lib)
+
+    def test_delete_missing_library_returns_404(self):
+        with patch.dict(
+            "sys.modules", {"library.models": self._fake_models_module(None)}
+        ), patch("library.api.views.delete_library_cascade") as mock_cascade:
+            response = self.client.delete("/library/api/libraries/nope/")
+
+        self.assertEqual(response.status_code, 404)
+        mock_cascade.assert_not_called()
+
+
+class CollectionApiDeleteTests(TestCase):
+    """DELETE on the collection endpoint cascades Items/Chunks/Images."""
+
+    def setUp(self):
+        self.user = User.objects.create_user(username="op", password="pw")
+        self.client = APIClient()
+        self.client.force_authenticate(user=self.user)
+
+    def test_delete_uses_shared_cascade(self):
+        col = SimpleNamespace(uid="col-1", name="Default")
+        with patch.dict(
+            "sys.modules",
+            {"library.models": SimpleNamespace(Collection=_fake_node_cls(col))},
+        ), patch(
+            "library.api.views.delete_collection_cascade",
+            return_value={
+                "collection_uid": "col-1",
+                "name": "Default",
+                "item_count": 2,
+                "item_s3_keys": [],
+            },
+        ) as mock_cascade:
+            response = self.client.delete("/library/api/collections/col-1/")
+
+        self.assertEqual(response.status_code, 204)
+        mock_cascade.assert_called_once_with(col)
+
+    def test_delete_missing_collection_returns_404(self):
+        with patch.dict(
+            "sys.modules",
+            {"library.models": SimpleNamespace(Collection=_fake_node_cls(None))},
+        ), patch("library.api.views.delete_collection_cascade") as mock_cascade:
+            response = self.client.delete("/library/api/collections/nope/")
+
+        self.assertEqual(response.status_code, 404)
+        mock_cascade.assert_not_called()
+
+
+class ItemApiDeleteTests(TestCase):
+    """DELETE on the item endpoint cascades Chunks/Images/embeddings."""
+
+    def setUp(self):
+        self.user = User.objects.create_user(username="op", password="pw")
+        self.client = APIClient()
+        self.client.force_authenticate(user=self.user)
+
+    def test_delete_uses_shared_cascade(self):
+        item = SimpleNamespace(uid="item-1", title="Doc")
+        with patch.dict(
+            "sys.modules",
+            {"library.models": SimpleNamespace(Item=_fake_node_cls(item))},
+        ), patch(
+            "library.api.views.delete_item_cascade",
+            return_value={"item_uid": "item-1", "s3_key": ""},
+        ) as mock_cascade:
+            response = self.client.delete("/library/api/items/item-1/")
+
+        self.assertEqual(response.status_code, 204)
+        mock_cascade.assert_called_once_with("item-1")
+
+    def test_delete_missing_item_returns_404(self):
+        with patch.dict(
+            "sys.modules",
+            {"library.models": SimpleNamespace(Item=_fake_node_cls(None))},
+        ), patch("library.api.views.delete_item_cascade") as mock_cascade:
+            response = self.client.delete("/library/api/items/nope/")
+
+        self.assertEqual(response.status_code, 404)
+        mock_cascade.assert_not_called()

mnemosyne/library/tests/test_managed_by.py

@@ -70,8 +70,9 @@ class _FakeLibrary:
     class DoesNotExist(Exception):
         pass
 
-    existing = None          # what nodes.get(name=...) returns
+    existing = None          # what find_library_by_name_ci returns
     instances = []           # constructor kwargs, in order
+    save_raises = None       # exception instance save() should raise, if any
 
     def __init__(self, **kwargs):
         type(self).instances.append(kwargs)
@@ -81,16 +82,20 @@ class _FakeLibrary:
         self.created_at = None
 
     def save(self):
+        if type(self).save_raises is not None:
+            raise type(self).save_raises
         return self
 
-    class _Nodes:
-        @staticmethod
-        def get(**kwargs):
-            if _FakeLibrary.existing is None:
-                raise _FakeLibrary.DoesNotExist()
+
+def _fake_find_ci(name):
+    _FakeLibrary.ci_queries.append(name)
     return _FakeLibrary.existing
 
-    nodes = _Nodes()
+
+def _fake_models_module():
+    return SimpleNamespace(
+        Library=_FakeLibrary, find_library_by_name_ci=_fake_find_ci
+    )
 
 
 class LibraryCreateStampingTests(TestCase):
@@ -101,16 +106,15 @@ class LibraryCreateStampingTests(TestCase):
         self.client = APIClient()
         _FakeLibrary.existing = None
         _FakeLibrary.instances = []
+        _FakeLibrary.save_raises = None
+        _FakeLibrary.ci_queries = []
 
-    def _post(self, token=None):
+    def _post(self, token=None, name="Docs"):
         self.client.force_authenticate(user=self.user, token=token)
-        with patch.dict(
-            "sys.modules",
-            {"library.models": SimpleNamespace(Library=_FakeLibrary)},
-        ):
+        with patch.dict("sys.modules", {"library.models": _fake_models_module()}):
             return self.client.post(
                 "/library/api/libraries/",
-                {"name": "Docs", "library_type": "technical"},
+                {"name": name, "library_type": "technical"},
                 format="json",
             )
 
@@ -129,7 +133,7 @@ class LibraryCreateStampingTests(TestCase):
 
     def test_duplicate_name_returns_409_name_conflict(self):
         _FakeLibrary.existing = SimpleNamespace(
-            uid="lib-old", managed_by_display="Daedalus"
+            uid="lib-old", name="Docs", managed_by_display="Daedalus"
         )
         response = self._post(token=UserToken(name="Spelunker"))
 
@@ -141,15 +145,39 @@ class LibraryCreateStampingTests(TestCase):
         self.assertEqual(body["managed_by"], "Daedalus")
         self.assertEqual(_FakeLibrary.instances, [])
 
+    def test_duplicate_check_is_case_insensitive(self):
+        """A case-variant name 409s and reports the existing spelling."""
+        _FakeLibrary.existing = SimpleNamespace(
+            uid="lib-old", name="Amazon Connect", managed_by_display="Spelunker"
+        )
+        response = self._post(token=None, name="amazon connect")
+
+        self.assertEqual(response.status_code, 409)
+        # The lookup received the posted name (case-folding happens in
+        # Cypher), and the response names the existing spelling.
+        self.assertEqual(_FakeLibrary.ci_queries, ["amazon connect"])
+        self.assertIn("Amazon Connect", response.json()["detail"])
+        self.assertEqual(_FakeLibrary.instances, [])
+
     def test_duplicate_of_unmanaged_reports_null_manager(self):
         _FakeLibrary.existing = SimpleNamespace(
-            uid="lib-old", managed_by_display=""
+            uid="lib-old", name="Docs", managed_by_display=""
         )
         response = self._post(token=None)
 
         self.assertEqual(response.status_code, 409)
         self.assertIsNone(response.json()["managed_by"])
 
+    def test_save_race_returns_409_not_500(self):
+        """An exact-name twin landing between pre-check and save 409s."""
+        from neomodel.exceptions import UniqueProperty
+
+        _FakeLibrary.save_raises = UniqueProperty("name")
+        response = self._post(token=None)
+
+        self.assertEqual(response.status_code, 409)
+        self.assertEqual(response.json()["code"], "name_conflict")
+
 
 class WorkspaceStatusSerializerManagedByTests(TestCase):
     """The workspace status payload carries ``managed_by`` (nullable)."""

mnemosyne/library/tests/test_workspaces_api.py

@@ -7,6 +7,9 @@ search scoping) require Neo4j and are validated by the manual end-to-end
 test plan, not these unit tests.
 """
 
+from unittest.mock import Mock, patch
+
+from django.contrib.auth.models import User
 from django.test import TestCase
 from rest_framework.test import APIClient
 
@@ -208,3 +211,85 @@ class WorkspaceEndpointAuthTests(TestCase):
     def test_workspace_delete_requires_auth(self):
         response = self.client.delete("/library/api/workspaces/ws_a/")
         self.assertIn(response.status_code, [401, 403])
+
+
+class WorkspaceDeleteOwnershipTests(TestCase):
+    """DELETE must never report success for a library it did not delete.
+
+    A false 204 is how libraries get orphaned: Daedalus records the
+    workspace as cleaned up and drops its own row, while the Library node
+    survives holding its globally-unique name forever — which then blocks
+    ever recreating a workspace under that name.
+    """
+
+    def setUp(self):
+        self.client = APIClient()
+        self.user = User.objects.create_user(
+            username="owner", password="pw"  # noqa: S106 — test credential
+        )
+        self.client.force_authenticate(user=self.user)
+
+    def _library(self, owner_username):
+        lib = Mock()
+        lib.uid = "lib_1"
+        lib.owner_username = owner_username
+        return lib
+
+    def test_absent_library_still_returns_204(self):
+        """Genuine idempotency is preserved — nothing exists, nothing to do."""
+        from library.models import Library
+
+        with patch(
+            "neomodel.sync_.match.NodeSet.get",
+            side_effect=Library.DoesNotExist("nope"),
+        ):
+            response = self.client.delete("/library/api/workspaces/ws_gone/")
+
+        self.assertEqual(response.status_code, 204)
+
+    def test_unowned_library_returns_409_and_is_not_deleted(self):
+        from library.models import Library
+
+        lib = self._library("someone_else")
+        with (
+            patch("neomodel.sync_.match.NodeSet.get", return_value=lib),
+            patch(
+                "library.api.workspaces.delete_library_cascade"
+            ) as cascade,
+        ):
+            response = self.client.delete("/library/api/workspaces/ws_a/")
+
+        self.assertEqual(response.status_code, 409)
+        self.assertEqual(response.json()["code"], "owner_conflict")
+        cascade.assert_not_called()
+
+    def test_owned_library_is_deleted(self):
+        from library.models import Library
+
+        lib = self._library("owner")
+        with (
+            patch("neomodel.sync_.match.NodeSet.get", return_value=lib),
+            patch(
+                "library.api.workspaces.delete_library_cascade",
+                return_value={
+                    "library_uid": "lib_1",
+                    "name": "Assistant",
+                    "item_count": 0,
+                    "orphans_deleted": 0,
+                },
+            ) as cascade,
+        ):
+            response = self.client.delete("/library/api/workspaces/ws_a/")
+
+        self.assertEqual(response.status_code, 204)
+        cascade.assert_called_once_with(lib)
+
+    def test_unowned_library_get_still_looks_absent(self):
+        """Ownership must stay opaque on reads — 404, not 409."""
+        from library.models import Library
+
+        lib = self._library("someone_else")
+        with patch("neomodel.sync_.match.NodeSet.get", return_value=lib):
+            response = self.client.get("/library/api/workspaces/ws_a/")
+
+        self.assertEqual(response.status_code, 404)

mnemosyne/library/views.py

@@ -460,7 +460,11 @@ def collection_delete(request, uid):
 
     if request.method == "POST":
         name = col.name
-        col.delete()
+        # Shared cascade so Items/Chunks/Images go too — a bare
+        # col.delete() would leak them.
+        from .services.library_delete import delete_collection_cascade
+
+        delete_collection_cascade(col)
         messages.success(request, f'Collection "{name}" deleted.')
         return redirect("library:library-list")
     return render(
@@ -648,7 +652,11 @@ def item_delete(request, uid):
 
     if request.method == "POST":
         title = item.title
-        item.delete()
+        # Shared cascade so Chunks/Images/embeddings go too — a bare
+        # item.delete() would leak them.
+        from .services.library_delete import delete_item_cascade
+
+        delete_item_cascade(item.uid)
         messages.success(request, f'Item "{title}" deleted.')
         return redirect("library:library-list")
     return render(request, "library/item_confirm_delete.html", {"item": item})