From 806aae06da109ffc5d2f509e13f137b785a76bae Mon Sep 17 00:00:00 2001 From: Devil Date: Mon, 5 Oct 2026 20:18:16 -0400 Subject: [PATCH] fix: close local Qdrant persistence before collection deletion --- CHANGELOG.md | 2 +- daem0nmcp/api/v7/portable_projections.py | 13 +- daem0nmcp/qdrant_store.py | 3 +- daem0nmcp/retrieval/dense_generation_gc.py | 15 ++- daem0nmcp/retrieval/dense_projection.py | 21 +--- daem0nmcp/retrieval/providers.py | 17 +++ tests/api_v7/test_portable_projections.py | 140 +++++++++++++++++++++ tests/test_retrieval_dense_projection.py | 71 +++++++++++ tests/test_vectors.py | 52 ++++++++ 9 files changed, 300 insertions(+), 34 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index bf7c9f1..cc0412a 100644 --- a/CHANGELOG.md +++ b/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. diff --git a/daem0nmcp/api/v7/portable_projections.py b/daem0nmcp/api/v7/portable_projections.py index 68a7f35..7b3879e 100644 --- a/daem0nmcp/api/v7/portable_projections.py +++ b/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(" diff --git a/daem0nmcp/qdrant_store.py b/daem0nmcp/qdrant_store.py index 9eab374..cbba046 100644 --- a/daem0nmcp/qdrant_store.py +++ b/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: diff --git a/daem0nmcp/retrieval/dense_generation_gc.py b/daem0nmcp/retrieval/dense_generation_gc.py index e17b77f..eeb08fd 100644 --- a/daem0nmcp/retrieval/dense_generation_gc.py +++ b/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 diff --git a/daem0nmcp/retrieval/dense_projection.py b/daem0nmcp/retrieval/dense_projection.py index 2932190..3f3c650 100644 --- a/daem0nmcp/retrieval/dense_projection.py +++ b/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, diff --git a/daem0nmcp/retrieval/providers.py b/daem0nmcp/retrieval/providers.py index ea12295..eee05fe 100644 --- a/daem0nmcp/retrieval/providers.py +++ b/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", ] diff --git a/tests/api_v7/test_portable_projections.py b/tests/api_v7/test_portable_projections.py index 7ab99fe..87c08bc 100644 --- a/tests/api_v7/test_portable_projections.py +++ b/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) diff --git a/tests/test_retrieval_dense_projection.py b/tests/test_retrieval_dense_projection.py index 04c233e..7169d7f 100644 --- a/tests/test_retrieval_dense_projection.py +++ b/tests/test_retrieval_dense_projection.py @@ -4,9 +4,11 @@ import asyncio import copy +import gc import hashlib import importlib.util import json +import os import shutil import sqlite3 import tempfile @@ -741,6 +743,75 @@ def test_replacing_local_orphan_staging_discards_its_persisted_points(self): shutil.rmtree(qdrant_path) self.assertFalse(qdrant_path.exists()) + @unittest.skipUnless(os.name == "nt", "native Windows storage ownership") + @unittest.skipUnless( + importlib.util.find_spec("qdrant_client"), + "optional local Qdrant is unavailable", + ) + def test_retired_local_generation_releases_storage_without_cyclic_gc(self): + from qdrant_client import QdrantClient + + from daem0nmcp.retrieval.dense_generation_gc import ( + DenseGenerationGarbageCollector, + ) + from daem0nmcp.retrieval.dense_projection import DenseProjectionBuilder + + qdrant_path = self.database_path.parent / "retired-local-qdrant" + client = QdrantClient(path=str(qdrant_path)) + builder = DenseProjectionBuilder( + self.connection, + provider_key="local", + model_id="deterministic-test-model", + dimension=3, + encoder=self.encoder, + client=client, + collection_prefix="test-dense", + clock_us=lambda: 500, + ) + gc_was_enabled = gc.isenabled() + gc.disable() + try: + generations = [] + for index, suffix in enumerate(("a", "b", "c"), start=1): + self._append(suffix, f"native generation {index}") + generations.append(builder.rebuild(WORKSPACE_ID)) + + result = DenseGenerationGarbageCollector( + self.connection, + client=client, + clock_us=lambda: 501, + token_factory=lambda: "native-windows-gc-claim", + ).run_slice(max_generations=2) + self.assertEqual( + [(1, "succeeded")], + [(item.generation, item.status) for item in result], + ) + retired = generations[0].collection_name + self.assertFalse(client.collection_exists(retired)) + self.assertFalse((qdrant_path / "collection" / retired).exists()) + self.assertEqual( + [(2, "ready"), (3, "active")], + [ + tuple(row) + for row in self.connection.execute( + "SELECT generation,status FROM projection_manifests " + "WHERE projection_name='dense' ORDER BY generation" + ) + ], + ) + for generation in generations[1:]: + self.assertTrue(client.collection_exists(generation.collection_name)) + builder.close() + client.close() + released_path = qdrant_path.with_name("released-local-qdrant") + qdrant_path.rename(released_path) + shutil.rmtree(released_path) + finally: + builder.close() + client.close() + if gc_was_enabled: + gc.enable() + def test_cosine_rounding_does_not_admit_corrupt_or_malformed_vectors(self): from daem0nmcp.retrieval.vector_validation import cosine_vectors_match diff --git a/tests/test_vectors.py b/tests/test_vectors.py index f5420a9..6346e8a 100644 --- a/tests/test_vectors.py +++ b/tests/test_vectors.py @@ -1,5 +1,8 @@ """Tests for the vector embeddings module.""" +import gc +import os +import shutil from importlib.util import find_spec from unittest.mock import MagicMock, patch @@ -180,3 +183,52 @@ def test_returns_configured_dimension(self): from daem0nmcp.vectors import get_dimension assert get_dimension() == settings.embedding_dimension + + +@pytest.mark.skipif(os.name != "nt", reason="native Windows storage ownership") +def test_local_dimension_replacement_releases_storage_without_cyclic_gc(tmp_path): + qdrant = pytest.importorskip("qdrant_client") + from daem0nmcp.qdrant_store import QdrantVectorStore + + class ReconfiguredStore(QdrantVectorStore): + EMBEDDING_DIMENSION = 2 + + storage = tmp_path / "qdrant" + seed = qdrant.QdrantClient(path=str(storage)) + store = None + gc_was_enabled = gc.isenabled() + gc.disable() + try: + collections = ( + QdrantVectorStore.COLLECTION_MEMORIES, + QdrantVectorStore.COLLECTION_CODE, + ) + for collection_name in collections: + seed.create_collection( + collection_name=collection_name, + vectors_config=qdrant.models.VectorParams( + size=3, distance=qdrant.models.Distance.COSINE + ), + ) + seed.upsert( + collection_name=collection_name, + points=[qdrant.models.PointStruct(id=1, vector=[1.0, 0.0, 0.0])], + wait=True, + ) + seed.close() + store = ReconfiguredStore(path=str(storage)) + for collection_name in collections: + info = store.client.get_collection(collection_name) + assert info.config.params.vectors.size == 2 + count = store.client.count(collection_name=collection_name, exact=True) + assert count.count == 0 + store.close() + released_storage = storage.with_name("released-qdrant") + storage.rename(released_storage) + shutil.rmtree(released_storage) + finally: + if store is not None: + store.close() + seed.close() + if gc_was_enabled: + gc.enable()