diff --git a/.superpowers/sdd/2026-08-08-internal-qdrant-ollama/task-5-report.md b/.superpowers/sdd/2026-08-08-internal-qdrant-ollama/task-5-report.md new file mode 100644 index 00000000..9608fec6 --- /dev/null +++ b/.superpowers/sdd/2026-08-08-internal-qdrant-ollama/task-5-report.md @@ -0,0 +1,98 @@ +# Task 5 Report — Implement the Qdrant VectorStore adapter + +## Status + +Implemented on 2026-08-08 in `/Users/mp/projects/ThothII/.worktrees/git-workspace-registry`. + +## RED evidence + +Command: + +```bash +cd harness +./.venv/bin/pytest tests/test_qdrant_vector_store.py tests/test_vector_port_contract.py -q +``` + +Observed before implementation: + +- exit code `2` +- collection failed during import because the adapter did not exist yet + +Representative failures: + +- `ModuleNotFoundError: No module named 'tht.adapters.vector.qdrant'` + +## GREEN evidence + +Focused behavior suite: + +```bash +cd harness +./.venv/bin/pytest tests/test_qdrant_vector_store.py tests/test_vector_port_contract.py -q +``` + +- exit code `0` +- `31 passed, 1 warning` + +Touched-file lint: + +```bash +cd harness +./.venv/bin/ruff check tht/adapters/vector/qdrant.py tht/adapters/vector/__init__.py \ + tht/ports/vector.py tht/vectorstore/records.py tht/vectorstore/store.py \ + tests/test_qdrant_vector_store.py tests/test_vector_port_contract.py +``` + +- exit code `0` +- `All checks passed!` + +Patch hygiene: + +```bash +git diff --check +``` + +- exit code `0` + +## What changed + +- added `QdrantVectorStore` with direct `requests`-based REST calls for: + - `GET /collections/{collection}` + - `PUT /collections/{collection}` + - `PUT /collections/{collection}/index` + - `PUT /collections/{collection}/points?wait=true` + - `POST /collections/{collection}/points/query` + - `POST /collections/{collection}/points/scroll` + - `POST /collections/{collection}/points/delete?wait=true` +- implemented idempotent collection provisioning for `1024` dimensions and `Cosine` distance +- created deterministic UUIDv5 point IDs from workspace, semantic kind, and canonical record key +- preserved canonical record identity and only upserted/deleted points matching the exact workspace + and generation filters +- added Qdrant payload helpers so stored payloads carry: + - `workspace_id` + - grouped semantic `kind` (`schema`, `evidence`, `memory`) + - original `record_kind` + - canonical `record_key` + - `content_hash` + - existing Thoth metadata fields +- mapped Qdrant payloads back into existing `VectorHit` objects without losing the original + Thoth kind +- exported the new adapter from the public vector adapter package and added focused contract tests +- sanitized timeout and malformed-response failures so CLI-facing callers do not leak raw endpoint + details + +## Self-review + +- confirmed collection mismatch fails without any delete/recreate path +- confirmed every query/scroll/delete operation includes a workspace filter +- confirmed the adapter never deletes or rewrites unrelated Qdrant points +- added keyword payload indexes for all filter-critical fields used here, including `document_id` + for exact Evidence filtering + +## Concerns + +- the requested `adversarial-review` skill could not run its full external reviewer flow in this + environment because the skill’s referenced `brain/` files are missing at + `/Users/mp/.agents/skills/adversarial-review`; I performed a manual adversarial self-review + instead +- the focused suite still emits one pre-existing warning from `testcontainers.postgres` diff --git a/harness/tests/test_qdrant_vector_store.py b/harness/tests/test_qdrant_vector_store.py new file mode 100644 index 00000000..3414df41 --- /dev/null +++ b/harness/tests/test_qdrant_vector_store.py @@ -0,0 +1,312 @@ +import json +from uuid import NAMESPACE_URL, uuid5 + +import pytest +import requests + +from tht.adapters.vector.qdrant import QdrantVectorStore, point_id +from tht.ports.vector import VectorStoreError, VectorWriteRecord +from tht.vectorstore.records import VectorRecord + + +class FakeResponse: + def __init__(self, status_code: int, payload=None, text: str | None = None): + self.status_code = status_code + self._payload = payload + self.text = text if text is not None else ( + "" if payload is None else json.dumps(payload) + ) + + @property + def ok(self) -> bool: + return 200 <= self.status_code < 300 + + def json(self): + if isinstance(self._payload, Exception): + raise self._payload + return self._payload + + +class FakeQdrantHttp: + def __init__(self, *, dimension=1024, distance="Cosine"): + self.dimension = dimension + self.distance = distance + self.collection = None + self.payload_indexes: set[str] = set() + self.points: dict[str, dict] = {} + self.calls: list[tuple[str, str, dict | None]] = [] + self.fail_request: Exception | None = None + self.malformed_query = False + self.malformed_scroll = False + + def request(self, method, url, *, json=None, timeout=None): + self.calls.append((method, url, json)) + if self.fail_request is not None: + raise self.fail_request + + path = url.split("://", 1)[-1].split("/", 1)[-1] + path = "/" + path.split("?", 1)[0] + + if method == "GET" and path == "/collections/workspace-semantic": + if self.collection is None: + return FakeResponse(404, {"status": "error"}) + return FakeResponse(200, { + "result": { + "config": { + "params": { + "vectors": {"size": self.dimension, "distance": self.distance} + } + }, + "payload_schema": { + field: {"data_type": "keyword"} for field in sorted(self.payload_indexes) + }, + } + }) + + if method == "PUT" and path == "/collections/workspace-semantic": + self.collection = json + self.dimension = json["vectors"]["size"] + self.distance = json["vectors"]["distance"] + return FakeResponse(200, {"status": "ok"}) + + if method == "PUT" and path == "/collections/workspace-semantic/index": + self.payload_indexes.add(json["field_name"]) + return FakeResponse(200, {"status": "ok"}) + + if method == "PUT" and path == "/collections/workspace-semantic/points": + for point in json["points"]: + self.points[point["id"]] = point + return FakeResponse(200, {"result": {"status": "acknowledged"}}) + + if method == "POST" and path == "/collections/workspace-semantic/points/query": + if self.malformed_query: + return FakeResponse(200, {"result": {"points": "nope"}}) + wanted = _match_points(self.points.values(), json["filter"]) + scored = sorted( + ( + { + "id": point["id"], + "score": point.get("score", 0.9), + "payload": point["payload"], + } + for point in wanted + ), + key=lambda point: (-point["score"], point["payload"]["record_key"]), + ) + return FakeResponse(200, {"result": {"points": scored[: json["limit"]]}}) + + if method == "POST" and path == "/collections/workspace-semantic/points/scroll": + if self.malformed_scroll: + return FakeResponse(200, {"result": {"points": "bad"}}) + wanted = sorted( + _match_points(self.points.values(), json["filter"]), + key=lambda point: point["payload"]["record_key"], + ) + return FakeResponse(200, {"result": {"points": wanted}}) + + if method == "POST" and path == "/collections/workspace-semantic/points/delete": + doomed = [point["id"] for point in _match_points(self.points.values(), json["filter"])] + for point_id_value in doomed: + self.points.pop(point_id_value, None) + return FakeResponse(200, {"result": {"status": "acknowledged"}}) + + raise AssertionError((method, path, json)) + + +def _match_points(points, flt): + matches = [] + must = flt["must"] + for point in points: + payload = point["payload"] + if all(_match_clause(payload, clause) for clause in must): + matches.append(point) + return matches + + +def _match_clause(payload, clause): + key = clause["key"] + match = clause["match"] + if "value" in match: + return payload.get(key) == match["value"] + if "any" in match: + return payload.get(key) in set(match["any"]) + raise AssertionError(clause) + + +def _write_record(record_id: str, kind: str, *, metadata=None): + return VectorWriteRecord( + record=VectorRecord( + id=record_id, + kind=kind, + ref=f"ref:{record_id}", + title=f"title:{record_id}", + content=f"content:{record_id}", + metadata=metadata or {}, + ), + embedding=[0.1] * 1024, + content_hash="sha256:" + "a" * 64, + ) + + +def _store(fake: FakeQdrantHttp) -> QdrantVectorStore: + return QdrantVectorStore( + base_url="http://qdrant:6333", + collection="workspace-semantic", + workspace_id="demo", + expected_dimension=1024, + request=fake.request, + ) + + +def test_point_id_is_deterministic_uuidv5(): + assert point_id("demo", "memory", "memory:1") == str( + uuid5(NAMESPACE_URL, "thothii:demo:memory:memory:1") + ) + + +def test_upsert_creates_collection_and_keyword_indexes_idempotently(): + fake = FakeQdrantHttp() + store = _store(fake) + + assert store.upsert("memory", [_write_record("memory:1", "memory")]) == 1 + assert store.upsert("memory", [_write_record("memory:1", "memory")]) == 1 + + creates = [call for call in fake.calls if call[0] == "PUT" and call[1].endswith("/collections/workspace-semantic")] + assert len(creates) == 1 + assert creates[0][2] == {"vectors": {"size": 1024, "distance": "Cosine"}} + assert fake.payload_indexes == { + "content_hash", + "document_id", + "kind", + "record_key", + "record_kind", + "vector_generation", + "workspace_id", + } + + +def test_upsert_refuses_collection_dimension_or_distance_mismatch_without_recreating(): + fake = FakeQdrantHttp(dimension=384, distance="Dot") + fake.collection = {"vectors": {"size": 384, "distance": "Dot"}} + store = _store(fake) + + with pytest.raises(VectorStoreError, match="Qdrant collection configuration mismatch"): + store.upsert("memory", [_write_record("memory:1", "memory")]) + + creates = [call for call in fake.calls if call[0] == "PUT" and call[1].endswith("/collections/workspace-semantic")] + assert creates == [] + + +@pytest.mark.parametrize( + ("record", "semantic_kind"), + [ + (_write_record("schema_column:patients.id", "schema_column"), "schema"), + ( + _write_record( + "demo:gen:11111111111111111111111111111111:chunk:1", + "evidence", + metadata={ + "workspace_id": "demo", + "vector_generation": "gen:11111111111111111111111111111111", + "document_id": "doc:abc", + }, + ), + "evidence", + ), + (_write_record("memory:1", "memory"), "memory"), + ], +) +def test_upsert_serializes_qdrant_point_payloads(record, semantic_kind): + fake = FakeQdrantHttp() + store = _store(fake) + + store.upsert("memory" if semantic_kind == "memory" else "evidence" if semantic_kind == "evidence" else "schema_records", [record]) + + point = next(iter(fake.points.values())) + assert point["id"] == point_id("demo", semantic_kind, record.record.id) + assert point["vector"] == record.embedding + assert point["payload"]["workspace_id"] == "demo" + assert point["payload"]["kind"] == semantic_kind + assert point["payload"]["record_kind"] == record.record.kind + assert point["payload"]["record_key"] == record.record.id + assert point["payload"]["content_hash"] == record.content_hash + + +def test_search_filters_by_workspace_and_allowed_record_kinds(): + fake = FakeQdrantHttp() + store = _store(fake) + store.upsert("memory", [_write_record("memory:1", "memory")]) + other = next(iter(fake.points.values())).copy() + other["id"] = point_id("other", "memory", "memory:2") + other["payload"] = {**other["payload"], "workspace_id": "other", "record_key": "memory:2"} + fake.points[other["id"]] = other + solved = next(iter(fake.points.values())).copy() + solved["id"] = point_id("demo", "memory", "solved:1") + solved["payload"] = {**solved["payload"], "record_key": "solved:1", "record_kind": "solved_question"} + fake.points[solved["id"]] = solved + + hits = store.search(["memory"], [0.2] * 1024, limit=5, kinds=["memory"]) + + assert [hit.id for hit in hits] == ["memory:1"] + query_call = next(call for call in fake.calls if call[0] == "POST" and call[1].endswith("/points/query?wait=true") is False and call[1].endswith("/points/query")) + assert query_call[2]["filter"] == { + "must": [ + {"key": "workspace_id", "match": {"value": "demo"}}, + {"key": "record_kind", "match": {"any": ["memory"]}}, + ] + } + + +def test_existing_hashes_health_and_exact_generation_inventory_and_delete(): + fake = FakeQdrantHttp() + store = _store(fake) + generation = "gen:" + "1" * 32 + keep = "gen:" + "2" * 32 + store.upsert("evidence", [ + _write_record( + f"demo:{generation}:chunk:1", + "evidence", + metadata={"workspace_id": "demo", "vector_generation": generation, "document_id": "doc:1"}, + ), + _write_record( + f"demo:{keep}:chunk:2", + "evidence", + metadata={"workspace_id": "demo", "vector_generation": keep, "document_id": "doc:2"}, + ), + ]) + + assert store.existing_hashes("evidence", ["evidence"]) == { + f"demo:{generation}:chunk:1": "sha256:" + "a" * 64, + f"demo:{keep}:chunk:2": "sha256:" + "a" * 64, + } + assert store.list_evidence_generations("evidence", "demo") == [generation, keep] + assert store.delete_generation("evidence", generation, "demo") == 1 + assert store.list_evidence_generations("evidence", "demo") == [keep] + + health = store.health() + assert health.ok is True + assert health.read_reachable is True + assert health.write_reachable is True + assert health.observed_dimensions == (1024,) + assert health.dimension_compatible is True + + +def test_sanitizes_timeout_and_malformed_responses(): + fake = FakeQdrantHttp() + store = _store(fake) + fake.fail_request = requests.Timeout("dial tcp 10.0.0.9:6333: i/o timeout") + + with pytest.raises(VectorStoreError, match="Qdrant request failed") as timeout: + store.search(["memory"], [0.2] * 1024, limit=1) + assert "10.0.0.9" not in str(timeout.value) + + fake.fail_request = None + store.upsert("memory", [_write_record("memory:1", "memory")]) + fake.malformed_query = True + with pytest.raises(VectorStoreError, match="Qdrant returned malformed query response"): + store.search(["memory"], [0.2] * 1024, limit=1) + + fake.malformed_query = False + fake.malformed_scroll = True + with pytest.raises(VectorStoreError, match="Qdrant returned malformed scroll response"): + store.existing_hashes("memory", ["memory"]) diff --git a/harness/tests/test_vector_port_contract.py b/harness/tests/test_vector_port_contract.py index db97831d..42852a52 100644 --- a/harness/tests/test_vector_port_contract.py +++ b/harness/tests/test_vector_port_contract.py @@ -3,14 +3,15 @@ from unittest.mock import MagicMock import pytest -from tht.adapters.vector.thoth_http import ThothHttpVectorStore from tht.adapters.vector.legacy_direct import LegacyDirectVectorStore +from tht.adapters.vector.qdrant import QdrantVectorStore +from tht.adapters.vector.thoth_http import ThothHttpVectorStore from tht.evidence.model import EvidenceDoc from tht.ports.vector import ( VectorHit, + VectorReadUnavailable, VectorRecord, VectorStore, - VectorReadUnavailable, VectorWriteRecord, VectorWriteUnavailable, ) @@ -158,11 +159,13 @@ def test_http_store_is_runtime_vector_store(): def test_vector_contract_is_exported_from_public_packages(): + from tht.adapters.vector import QdrantVectorStore as PublicQdrantStore from tht.adapters.vector import ThothHttpVectorStore as PublicHttpStore + from tht.ports import VectorReadUnavailable as PublicVectorReadUnavailable from tht.ports import VectorStore as PublicVectorStore from tht.ports import VectorWriteRecord as PublicVectorWriteRecord - from tht.ports import VectorReadUnavailable as PublicVectorReadUnavailable + assert PublicQdrantStore is QdrantVectorStore assert PublicHttpStore is ThothHttpVectorStore assert PublicVectorStore is VectorStore assert PublicVectorWriteRecord is VectorWriteRecord @@ -248,3 +251,24 @@ def test_legacy_direct_search_requires_a_strict_positive_integer_limit(limit): with pytest.raises(ValueError, match="positive integer"): store.search(["memory"], [0.1], limit=limit) + + +def test_qdrant_store_is_runtime_vector_store(): + store = QdrantVectorStore( + base_url="http://qdrant:6333", + collection="workspace-semantic", + workspace_id="demo", + expected_dimension=1024, + request=lambda *args, **kwargs: MagicMock( + ok=True, + status_code=200, + text='{"result":{"config":{"params":{"vectors":{"size":1024,"distance":"Cosine"}}},"payload_schema":{}}}', + json=lambda: { + "result": { + "config": {"params": {"vectors": {"size": 1024, "distance": "Cosine"}}}, + "payload_schema": {}, + } + }, + ), + ) + assert isinstance(store, VectorStore) diff --git a/harness/tht/adapters/vector/__init__.py b/harness/tht/adapters/vector/__init__.py index 4c6fa146..4a3421cb 100644 --- a/harness/tht/adapters/vector/__init__.py +++ b/harness/tht/adapters/vector/__init__.py @@ -2,6 +2,7 @@ from tht.adapters.vector.legacy_direct import LegacyDirectVectorStore from tht.adapters.vector.pgvector import PgVectorStore +from tht.adapters.vector.qdrant import QdrantVectorStore from tht.adapters.vector.thoth_http import ThothHttpVectorStore -__all__ = ["LegacyDirectVectorStore", "PgVectorStore", "ThothHttpVectorStore"] +__all__ = ["LegacyDirectVectorStore", "PgVectorStore", "QdrantVectorStore", "ThothHttpVectorStore"] diff --git a/harness/tht/adapters/vector/qdrant.py b/harness/tht/adapters/vector/qdrant.py new file mode 100644 index 00000000..52062982 --- /dev/null +++ b/harness/tht/adapters/vector/qdrant.py @@ -0,0 +1,363 @@ +"""Qdrant-backed vector store for one workspace-owned semantic collection.""" + +import re +from collections.abc import Callable +from uuid import NAMESPACE_URL, uuid5 + +import requests + +from tht.adapters.vector.pgvector import ( + COLLECTION_KINDS, + _collection, + _validate_collection_kinds, + _validate_known_kinds, +) +from tht.ports.vector import ( + VectorCapabilities, + VectorHealth, + VectorStoreError, + VectorWriteRecord, + require_positive_limit, +) +from tht.vectorstore.records import qdrant_payload, qdrant_semantic_kind +from tht.vectorstore.store import VectorHit, hit_from_metadata + +_GENERATION = re.compile(r"gen:[0-9a-f]{32}") +_WORKSPACE = re.compile(r"[a-z][a-z0-9_-]{0,63}") +_KEYWORD_INDEXES = ( + "content_hash", + "document_id", + "kind", + "record_key", + "record_kind", + "vector_generation", + "workspace_id", +) + + +def point_id(workspace_id: str, kind: str, record_key: str) -> str: + return str(uuid5(NAMESPACE_URL, f"thothii:{workspace_id}:{kind}:{record_key}")) + + +def _sanitize_exception(exc: Exception) -> str: + if isinstance(exc, requests.Timeout): + return "Qdrant request failed: timeout" + return f"Qdrant request failed: {type(exc).__name__}" + + +class QdrantVectorStore: + def __init__( + self, + *, + base_url: str, + collection: str, + workspace_id: str, + expected_dimension: int | None = None, + request: Callable[..., object] | None = None, + connect_timeout: float = 2.0, + read_timeout: float = 10.0, + ): + self._base_url = base_url.rstrip("/") + self._collection = collection + self._workspace_id = workspace_id + self._expected_dimension = expected_dimension + self._request = request or requests.request + self._timeout = (connect_timeout, read_timeout) + + @property + def capabilities(self) -> VectorCapabilities: + return VectorCapabilities( + search=True, + existing_hashes=True, + upsert=True, + metadata_filter=True, + delete_generation=True, + list_evidence_generations=True, + ) + + def health(self) -> VectorHealth: + try: + info = self._ensure_collection(strict=False) + except VectorStoreError as exc: + return VectorHealth( + ok=False, + detail=str(exc), + read_configured=True, + read_reachable=False, + read_detail=str(exc), + write_configured=True, + write_reachable=False, + write_detail=str(exc), + expected_dimension=self._expected_dimension, + ) + + dimensions = () + compatible = None + if info is not None: + dimension = info["config"]["params"]["vectors"]["size"] + dimensions = (dimension,) + compatible = ( + None if self._expected_dimension is None else dimensions == (self._expected_dimension,) + ) + return VectorHealth( + ok=compatible is not False, + read_configured=True, + read_reachable=True, + write_configured=True, + write_reachable=True, + expected_dimension=self._expected_dimension, + observed_dimensions=dimensions, + dimension_compatible=compatible, + ) + + def search( + self, + collections: list[str], + embedding: list[float], + *, + limit: int, + kinds: list[str] | None = None, + metadata_filter: dict[str, object] | None = None, + ) -> list[VectorHit]: + require_positive_limit(limit) + self._validate_embedding(embedding, query=True) + allowed_record_kinds = self._allowed_record_kinds(collections, kinds) + if not allowed_record_kinds: + return [] + filter_must = self._workspace_filter() + filter_must.append({"key": "record_kind", "match": {"any": allowed_record_kinds}}) + if metadata_filter is not None: + if set(metadata_filter) != {"vector_generation", "document_ids", "workspace_id"}: + raise VectorStoreError("Unsupported vector metadata filter") + generation = metadata_filter["vector_generation"] + document_ids = metadata_filter["document_ids"] + workspace_id = metadata_filter["workspace_id"] + if ( + not isinstance(generation, str) + or not isinstance(document_ids, list) + or not all(isinstance(item, str) for item in document_ids) + or not isinstance(workspace_id, str) + ): + raise VectorStoreError("Invalid vector metadata filter") + filter_must = [ + {"key": "workspace_id", "match": {"value": workspace_id}}, + {"key": "record_kind", "match": {"any": allowed_record_kinds}}, + {"key": "vector_generation", "match": {"value": generation}}, + {"key": "document_id", "match": {"any": document_ids}}, + ] + response = self._call( + "POST", + f"/collections/{self._collection}/points/query", + { + "vector": embedding, + "limit": limit, + "with_payload": True, + "filter": {"must": filter_must}, + }, + ) + points = response.get("result", {}).get("points") + if not isinstance(points, list): + raise VectorStoreError("Qdrant returned malformed query response") + hits = [self._hit_from_point(point) for point in points] + return sorted(hits, key=lambda hit: (-hit.similarity, hit.id))[:limit] + + def existing_hashes(self, collection: str, kinds: list[str]) -> dict[str, str]: + _collection("vectors", collection) + _validate_collection_kinds(collection, kinds) + points = self._scroll( + [ + *self._workspace_filter(), + {"key": "record_kind", "match": {"any": sorted(kinds)}}, + ] + ) + hashes: dict[str, str] = {} + for point in points: + payload = point.get("payload") + if not isinstance(payload, dict): + raise VectorStoreError("Qdrant returned malformed scroll response") + record_key = payload.get("record_key") + content_hash = payload.get("content_hash") + if not isinstance(record_key, str) or not isinstance(content_hash, str): + raise VectorStoreError("Qdrant returned malformed scroll response") + hashes[record_key] = content_hash + return hashes + + def upsert(self, collection: str, records: list[VectorWriteRecord]) -> int: + _collection("vectors", collection) + self._ensure_collection(strict=True) + points = [] + for write_record in records: + _validate_collection_kinds(collection, [write_record.record.kind]) + self._validate_embedding(write_record.embedding, query=False) + semantic_kind = qdrant_semantic_kind(write_record.record.kind) + points.append( + { + "id": point_id(self._workspace_id, semantic_kind, write_record.record.id), + "vector": write_record.embedding, + "payload": qdrant_payload( + write_record.record, + content_hash=write_record.content_hash, + workspace_id=self._workspace_id, + ), + } + ) + self._call( + "PUT", + f"/collections/{self._collection}/points?wait=true", + {"points": points}, + ) + return len(records) + + def delete_generation(self, collection: str, generation: str, workspace_id: str) -> int: + if collection != "evidence" or _GENERATION.fullmatch(generation) is None: + raise VectorStoreError("Only exact Evidence generations may be deleted") + if _WORKSPACE.fullmatch(workspace_id) is None: + raise VectorStoreError("Invalid Evidence workspace namespace") + before = len( + self._scroll( + [ + {"key": "workspace_id", "match": {"value": workspace_id}}, + {"key": "record_kind", "match": {"any": ["evidence"]}}, + {"key": "vector_generation", "match": {"value": generation}}, + ] + ) + ) + self._call( + "POST", + f"/collections/{self._collection}/points/delete?wait=true", + { + "filter": { + "must": [ + {"key": "workspace_id", "match": {"value": workspace_id}}, + {"key": "record_kind", "match": {"any": ["evidence"]}}, + {"key": "vector_generation", "match": {"value": generation}}, + ] + } + }, + ) + return before + + def list_evidence_generations(self, collection: str, workspace_id: str) -> list[str]: + if collection != "evidence": + raise VectorStoreError("Only exact Evidence generations may be listed") + if _WORKSPACE.fullmatch(workspace_id) is None: + raise VectorStoreError("Invalid Evidence workspace namespace") + points = self._scroll( + [ + {"key": "workspace_id", "match": {"value": workspace_id}}, + {"key": "record_kind", "match": {"any": ["evidence"]}}, + ] + ) + generations = { + payload["vector_generation"] + for point in points + if isinstance((payload := point.get("payload")), dict) + and isinstance(payload.get("vector_generation"), str) + and _GENERATION.fullmatch(payload["vector_generation"]) is not None + } + return sorted(generations) + + def _workspace_filter(self) -> list[dict]: + return [{"key": "workspace_id", "match": {"value": self._workspace_id}}] + + def _allowed_record_kinds( + self, collections: list[str], kinds: list[str] | None + ) -> list[str]: + selected: set[str] = set() + for collection in collections: + _collection("vectors", collection) + selected.update(COLLECTION_KINDS[collection]) + if kinds: + _validate_known_kinds(kinds) + selected &= set(kinds) + return sorted(selected) + + def _validate_embedding(self, embedding: list[float], *, query: bool) -> None: + if self._expected_dimension is not None and len(embedding) != self._expected_dimension: + raise VectorStoreError( + "Query embedding dimension does not match configured dimension" + if query + else "Embedding dimension does not match configured dimension" + ) + + def _ensure_collection(self, *, strict: bool) -> dict | None: + response = self._call("GET", f"/collections/{self._collection}", None, allow_missing=True) + if response is None: + if not strict: + return None + self._call( + "PUT", + f"/collections/{self._collection}", + {"vectors": {"size": self._expected_dimension or 1024, "distance": "Cosine"}}, + ) + for field_name in _KEYWORD_INDEXES: + self._call( + "PUT", + f"/collections/{self._collection}/index", + {"field_name": field_name, "field_schema": "keyword"}, + ) + response = self._call("GET", f"/collections/{self._collection}", None) + result = response.get("result") if isinstance(response, dict) else None + config = result.get("config", {}).get("params", {}).get("vectors") if isinstance(result, dict) else None + if not isinstance(config, dict): + raise VectorStoreError("Qdrant returned malformed collection response") + size = config.get("size") + distance = config.get("distance") + if ( + self._expected_dimension is not None + and (size != self._expected_dimension or distance != "Cosine") + ): + raise VectorStoreError("Qdrant collection configuration mismatch") + for field_name in _KEYWORD_INDEXES: + if field_name not in result.get("payload_schema", {}): + self._call( + "PUT", + f"/collections/{self._collection}/index", + {"field_name": field_name, "field_schema": "keyword"}, + ) + return result + + def _scroll(self, must: list[dict]) -> list[dict]: + response = self._call( + "POST", + f"/collections/{self._collection}/points/scroll", + {"with_payload": True, "limit": 10000, "filter": {"must": must}}, + ) + points = response.get("result", {}).get("points") + if not isinstance(points, list): + raise VectorStoreError("Qdrant returned malformed scroll response") + return points + + def _hit_from_point(self, point: dict) -> VectorHit: + payload = point.get("payload") + score = point.get("score") + if not isinstance(payload, dict) or not isinstance(score, (int, float)): + raise VectorStoreError("Qdrant returned malformed query response") + return hit_from_metadata(float(score), payload) + + def _call(self, method: str, path: str, payload: dict | None, allow_missing: bool = False) -> dict | None: + try: + response = self._request( + method, + f"{self._base_url}{path}", + json=payload, + timeout=self._timeout, + ) + except requests.RequestException as exc: + raise VectorStoreError(_sanitize_exception(exc)) from exc + if response.status_code == 404 and allow_missing: + return None + if not response.ok: + raise VectorStoreError(f"Qdrant request failed: HTTP {response.status_code}") + if response.status_code == 204 or not getattr(response, "text", ""): + return {} + try: + data = response.json() + except Exception as exc: + raise VectorStoreError("Qdrant returned malformed JSON response") from exc + if not isinstance(data, dict): + raise VectorStoreError("Qdrant returned malformed JSON response") + return data + + +__all__ = ["QdrantVectorStore", "point_id"] diff --git a/harness/tht/ports/vector.py b/harness/tht/ports/vector.py index 9fd1cd8d..e2bdef21 100644 --- a/harness/tht/ports/vector.py +++ b/harness/tht/ports/vector.py @@ -89,10 +89,11 @@ __all__ = [ "VectorCapabilities", "VectorHealth", "VectorHit", - "VectorRecord", "VectorReadUnavailable", + "VectorRecord", "VectorStore", "VectorStoreError", "VectorWriteRecord", "VectorWriteUnavailable", + "require_positive_limit", ] diff --git a/harness/tht/vectorstore/records.py b/harness/tht/vectorstore/records.py index d7d9e9b2..6f1615c3 100644 --- a/harness/tht/vectorstore/records.py +++ b/harness/tht/vectorstore/records.py @@ -17,6 +17,31 @@ class VectorRecord(BaseModel): metadata: dict = {} +def qdrant_semantic_kind(kind: str) -> str: + if kind in {"schema_table", "schema_column"}: + return "schema" + if kind in {"memory", "solved_question"}: + return "memory" + if kind == "evidence": + return "evidence" + raise ValueError(f"Unsupported vector kind: {kind}") + + +def qdrant_payload(record: VectorRecord, *, content_hash: str, workspace_id: str) -> dict: + semantic_kind = qdrant_semantic_kind(record.kind) + return { + "workspace_id": workspace_id, + "kind": semantic_kind, + "record_kind": record.kind, + "record_key": record.id, + "ref": record.ref, + "title": record.title, + "content": record.content, + "content_hash": content_hash, + **record.metadata, + } + + def split_markdown(text: str, max_chars: int) -> list[str]: """Spezza un markdown: intero se sta nel limite, altrimenti per heading '##', e in ultima istanza per accumulo greedy di righe.""" diff --git a/harness/tht/vectorstore/store.py b/harness/tht/vectorstore/store.py index f485e241..5849be27 100644 --- a/harness/tht/vectorstore/store.py +++ b/harness/tht/vectorstore/store.py @@ -42,7 +42,7 @@ def hit_from_metadata(similarity: float, metadata: dict | None) -> VectorHit: md = metadata or {} return VectorHit( id=md.get("record_key", ""), - kind=md.get("kind", ""), + kind=md.get("record_kind", md.get("kind", "")), ref=md.get("ref", ""), title=md.get("title", ""), content=md.get("content", ""),