fix: close local Qdrant persistence before collection deletion

CHANGELOG.md

@@ -35,7 +35,7 @@
 - Explore recall now labels enabled retention `RETENTION_NOT_APPLICABLE`, preserving its baseline evidence and token usage rather than implying alternate packing.
 - Schema 33 upgrades recover missing derived learning parents from complete validated canonical replay before strict backfill, preserving transactional rollback and fail-closed authority validation. Malformed record payloads surface the schema-upgrade error contract instead of raw key/type exceptions.
 - Entity recall preserves retryable lexical catchup when a generation publishes between retrieval and readiness inspection, avoiding terminal capability errors for evidence that becomes available immediately after the query. Same-generation indexed misses remain terminal.
-- Local dense projection cleanup closes staging collection persistence before deleting it, preventing leaked SQLite handles on failed or cancelled builds.
+- Local Qdrant collection deletion closes persistence before dropping ownership, including dense staging cleanup, retired-generation garbage collection, portable import cleanup and legacy dimension replacement. Windows storage files and directories are released immediately after normal shutdown rather than waiting for cyclic garbage collection.
 - Claude Code hooks remind instead of block. `pre_edit` adds a one-line gateway file-recall reminder; `stop` suggests direct replay-safe memory calls with counsel-challenge retry guidance. Neither hook writes memory, denies an edit, or keeps the agent running.
 - OpenCode edits are no longer gated either: the OpenCode plugin runs the same `pre_edit` hook, which now always allows the edit. The plugin does not yet show the reminder; that rework comes with the edit-bridge removal.
 - `install-claude-hooks` and `uninstall-claude-hooks` refuse to touch a `~/.claude/settings.json` that is not valid JSON, instead of replacing it.

daem0nmcp/api/v7/portable_projections.py

@@ -48,6 +48,7 @@
     create_qdrant_client,
     dense_manifest_details,
     dense_query_encoder_matches_contract,
+    qdrant_delete_collection,
 )
 from ...retrieval.vector_validation import (
     VECTOR_ATTESTATION_FORMAT,
@@ -1811,7 +1812,7 @@ def discard(
             if connection is not None:
                 active = _active_vector_collection(connection, self.collection_name)
             if active is None:
-                self.client.delete_collection(self.collection_name)
+                qdrant_delete_collection(self.client, self.collection_name)
                 if (
                     connection is not None
                     and self.session_id is not None
@@ -1867,7 +1868,7 @@ def _reclaim_stale_vector_artifacts(
         if active is not None:
             continue
         if client.collection_exists(str(artifact_name)):
-            client.delete_collection(str(artifact_name))
+            qdrant_delete_collection(client, str(artifact_name))
         if checkpoint is not None:
             checkpoint()
         connection.execute(
@@ -2322,7 +2323,7 @@ def point_factory(**values: object) -> object:
                     and _active_vector_collection(connection, collection_name) is None
                 ):
                     if client.collection_exists(collection_name):
-                        client.delete_collection(collection_name)
+                        qdrant_delete_collection(client, collection_name)
                     # A failed or unacknowledged deletion must leave its durable
                     # recovery record intact for the next finalization owner.
                     if (
@@ -2417,7 +2418,7 @@ def point_factory(**values: object) -> object:
         except PortableTransferError as exc:
             candidate.collection_name = temporary_collection
             raise PortableTransferError("VECTOR_REBUILD_REQUIRED") from exc
-        candidate.client.delete_collection(temporary_collection)
+        qdrant_delete_collection(candidate.client, temporary_collection)
     else:
         if checkpoint is not None:
             checkpoint()
@@ -2458,7 +2459,7 @@ def point_factory(**values: object) -> object:
                         checkpoint()
                 if promotion_offset is None:
                     break
-            candidate.client.delete_collection(temporary_collection)
+            qdrant_delete_collection(candidate.client, temporary_collection)
             candidate.collection_name = final_collection
         if (
             candidate.session_id is not None
@@ -2479,7 +2480,7 @@ def point_factory(**values: object) -> object:
     except Exception:
         if not final_exists:
             with suppress(Exception):
-                candidate.client.delete_collection(final_collection)
+                qdrant_delete_collection(candidate.client, final_collection)
         raise
     connection.execute(
         "CREATE TEMP TABLE IF NOT EXISTS portable_vector_refs("

daem0nmcp/qdrant_store.py

@@ -25,6 +25,7 @@
 )
 
 from .config import settings
+from .retrieval.providers import qdrant_delete_collection
 
 logger = logging.getLogger(__name__)
 
@@ -68,7 +69,7 @@ def _ensure_collections(self) -> None:
                         f"Collection {coll_name} has dimension {existing_dim}, "
                         f"expected {self.EMBEDDING_DIMENSION}. Recreating collection."
                     )
-                    self.client.delete_collection(coll_name)
+                    qdrant_delete_collection(self.client, coll_name)
                     collections.remove(coll_name)
 
             if coll_name not in collections:

daem0nmcp/retrieval/dense_generation_gc.py

@@ -9,9 +9,12 @@
 import time
 from collections.abc import Callable, Mapping
 from dataclasses import dataclass
-from typing import Protocol, cast
 
-from .providers import dense_manifest_details, qdrant_collection_exists
+from .providers import (
+    dense_manifest_details,
+    qdrant_collection_exists,
+    qdrant_delete_collection,
+)
 
 _WORKSPACE_ID = re.compile(r"^ws_[0-9a-f]{24}$")
 _OWNER_ID = re.compile(r"^[A-Za-z0-9._:-]{1,128}$")
@@ -34,10 +37,6 @@ def __init__(self, code: str) -> None:
         super().__init__(code)
 
 
-class _DenseCollectionClient(Protocol):
-    def delete_collection(self, collection_name: str) -> object: ...
-
-
 @dataclass(frozen=True, slots=True)
 class DenseGenerationReadLease:
     workspace_id: str
@@ -522,7 +521,7 @@ def __init__(
         ):
             raise ValueError("claim_owner is invalid")
         self.connection = connection
-        self.client = cast(_DenseCollectionClient, client)
+        self.client = client
         self._clock_us = clock_us
         self._claim_owner = claim_owner
         self._token_factory = token_factory or (lambda: secrets.token_hex(24))
@@ -706,7 +705,7 @@ def _execute(self, claim: _GCClaim) -> DenseGenerationGCResult:
                     claim, "DENSE_GENERATION_GC_CANCELLED", consume=False
                 )
             if exists:
-                self.client.delete_collection(claim.collection_name)
+                qdrant_delete_collection(self.client, claim.collection_name)
             if self._cancelled():
                 return self._record_failure(
                     claim, "DENSE_GENERATION_GC_CANCELLED", consume=False

daem0nmcp/retrieval/dense_projection.py

@@ -25,6 +25,7 @@
     dense_builder_contract,
     dense_manifest_details,
     qdrant_collection_exists,
+    qdrant_delete_collection,
 )
 from .vector_validation import (
     VECTOR_ATTESTATION_FORMAT,
@@ -727,7 +728,7 @@ def _cleanup_staging(
         try:
             client = self._get_client()
             if self._collection_exists(client, collection_name):
-                self._delete_collection(client, collection_name)
+                qdrant_delete_collection(client, collection_name)
         except Exception:
             pass
 
@@ -1119,22 +1120,6 @@ def _models_for_client(self, client: object) -> object | None:
     def _collection_exists(client: object, collection_name: str) -> bool:
         return qdrant_collection_exists(client, collection_name)
 
-    @staticmethod
-    def _delete_collection(client: object, collection_name: str) -> None:
-        if type(client).__module__.startswith("qdrant_client"):
-            from qdrant_client import QdrantClient
-            from qdrant_client.local.qdrant_local import QdrantLocal
-
-            if isinstance(client, QdrantClient) and isinstance(
-                client._client, QdrantLocal
-            ):
-                collection = client._client.collections.get(collection_name)
-                if collection is not None:
-                    # Local deletion drops the collection without closing SQLite;
-                    # client.close() cannot reach that persistence afterward.
-                    collection.close()
-        client.delete_collection(collection_name)
-
     def _qdrant_vector_config(self, client: object) -> object:
         models = self._models_for_client(client)
         if models is None:
@@ -1183,7 +1168,7 @@ def _create_collection(self, collection_name: str) -> None:
         exists = self._collection_exists(client, collection_name)
         self._check_cancelled()
         if exists:
-            self._delete_collection(client, collection_name)
+            qdrant_delete_collection(client, collection_name)
             self._check_cancelled()
         client.create_collection(
             collection_name=collection_name,

daem0nmcp/retrieval/providers.py

@@ -498,6 +498,22 @@ def create_qdrant_client(
     return factory(path=path)
 
 
+def qdrant_delete_collection(client: object, collection_name: str) -> None:
+    """Release owned local persistence before removing a provider collection."""
+
+    if type(client).__module__.startswith("qdrant_client"):
+        from qdrant_client import QdrantClient
+        from qdrant_client.local.qdrant_local import QdrantLocal
+
+        if isinstance(client, QdrantClient) and isinstance(client._client, QdrantLocal):
+            collection = client._client.collections.get(collection_name)
+            if collection is not None:
+                # Local deletion drops the collection without closing SQLite;
+                # client.close() cannot reach that persistence afterward.
+                collection.close()
+    client.delete_collection(collection_name)
+
+
 def qdrant_collection_exists(client: object, collection_name: str) -> bool:
     """Check collection presence across the supported Qdrant client range."""
 
@@ -1463,4 +1479,5 @@ def _result(
     "dense_manifest_details",
     "dense_point_id",
     "qdrant_collection_exists",
+    "qdrant_delete_collection",
 ]

tests/api_v7/test_portable_projections.py

@@ -3,8 +3,11 @@
 from __future__ import annotations
 
 import asyncio
+import gc
 import hashlib
 import json
+import os
+import shutil
 import sqlite3
 import threading
 from datetime import timedelta
@@ -879,6 +882,143 @@ def checkpoint() -> None:
     assert backend.collections == {}
 
 
+@pytest.mark.skipif(os.name != "nt", reason="native Windows storage ownership")
+@pytest.mark.parametrize(
+    "cleanup",
+    ("discard", "cancelled", "promote", "promote_existing", "promotion_cancelled"),
+)
+def test_local_vector_candidate_cleanup_releases_storage_without_cyclic_gc(
+    tmp_path: Path, cleanup: str
+):
+    qdrant = pytest.importorskip("qdrant_client")
+    from daem0nmcp.api.v7.portable_projections import activate_vector_candidate
+
+    fixture = _fixture(tmp_path / "target")
+    vectors = tmp_path / "vectors.jsonl"
+    vectors.write_bytes(b"")
+    prepared = PreparedImport(
+        session_id="ipt_" + "4" * 64,
+        database_path=tmp_path / "events.db",
+        manifest={
+            "event_count": 0,
+            "event_root_hash": hashlib.sha256().hexdigest(),
+            "vectors": _vector_metadata("portable-model"),
+        },
+        legacy_path=None,
+        vectors_path=vectors,
+        page_count=1,
+    )
+    config = _config()
+    config.qdrant_url = None
+    config.qdrant_api_key = None
+    config.qdrant_path = fixture.storage / "qdrant"
+    connection = sqlite3.connect(fixture.database)
+    candidate = None
+    checkpoints = 0
+
+    def cancel_staging() -> None:
+        nonlocal checkpoints
+        checkpoints += 1
+        if checkpoints == 4:
+            raise PortableTransferError("CANCELLED")
+
+    gc_was_enabled = gc.isenabled()
+    gc.disable()
+    try:
+        if cleanup == "cancelled":
+            with pytest.raises(PortableTransferError, match="CANCELLED"):
+                prepare_vector_candidate(
+                    prepared,
+                    connection,
+                    WORKSPACE_ID,
+                    config,
+                    local_capability_ready=True,
+                    checkpoint=cancel_staging,
+                )
+            assert not list((config.qdrant_path / "collection").iterdir())
+        else:
+            candidate, diagnostic = prepare_vector_candidate(
+                prepared,
+                connection,
+                WORKSPACE_ID,
+                config,
+                local_capability_ready=True,
+            )
+            assert candidate is not None
+            assert diagnostic is None
+            temporary_collection = candidate.collection_name
+            if cleanup == "discard":
+                candidate.discard(connection, WORKSPACE_ID)
+            else:
+                final_collection = str(
+                    dense_manifest_details(
+                        workspace_id=WORKSPACE_ID,
+                        provider_key=candidate.provider_key,
+                        generation=candidate.generation,
+                        model_id=candidate.model_id,
+                        dimension=candidate.dimension,
+                        collection_prefix=candidate.collection_prefix,
+                    )["collection_name"]
+                )
+                if cleanup == "promote_existing":
+                    candidate.client.create_collection(
+                        collection_name=final_collection,
+                        vectors_config=qdrant.models.VectorParams(
+                            size=3, distance=qdrant.models.Distance.COSINE
+                        ),
+                    )
+                if cleanup == "promotion_cancelled":
+                    final_seen = False
+
+                    def cancel_promotion() -> None:
+                        nonlocal final_seen
+                        if candidate.client.collection_exists(final_collection):
+                            if final_seen:
+                                raise PortableTransferError("CANCELLED")
+                            final_seen = True
+
+                    with pytest.raises(PortableTransferError, match="CANCELLED"):
+                        activate_vector_candidate(
+                            candidate,
+                            connection,
+                            WORKSPACE_ID,
+                            now_us=500,
+                            checkpoint=cancel_promotion,
+                        )
+                    assert not candidate.client.collection_exists(final_collection)
+                    assert not (
+                        config.qdrant_path / "collection" / final_collection
+                    ).exists()
+                    candidate.discard(connection, WORKSPACE_ID)
+                else:
+                    activate_vector_candidate(
+                        candidate, connection, WORKSPACE_ID, now_us=500
+                    )
+                    connection.commit()
+                    assert candidate.client.collection_exists(final_collection)
+                    assert not candidate.client.collection_exists(temporary_collection)
+                    manifests = connection.execute(
+                        "SELECT generation,status,row_count FROM projection_manifests "
+                        "WHERE workspace_id=? AND projection_name='dense'",
+                        (WORKSPACE_ID,),
+                    ).fetchall()
+                    assert manifests == [(1, "active", 0)]
+                    candidate.close()
+            assert not (
+                config.qdrant_path / "collection" / temporary_collection
+            ).exists()
+        connection.close()
+        released_storage = fixture.storage.with_name("released-storage")
+        fixture.storage.rename(released_storage)
+        shutil.rmtree(released_storage)
+    finally:
+        if candidate is not None:
+            candidate.close()
+        connection.close()
+        if gc_was_enabled:
+            gc.enable()
+
+
 def _point_field(point: object, name: str):
     if isinstance(point, dict):
         return point.get(name)