From 99e0024973db237c35cda600592119c121a0bdb5 Mon Sep 17 00:00:00 2001 From: mptyl Date: Tue, 11 Aug 2026 07:33:22 +0200 Subject: [PATCH] fix: harden task3 qdrant compatibility boundaries --- .../test/workspace-runtime-renderer.test.ts | 33 +--- harness/tests/qdrant_test_helpers.py | 163 ++++++++++++++++ harness/tests/test_adapter_factory.py | 9 + harness/tests/test_qdrant_cli_commands.py | 47 +++++ harness/tests/test_qdrant_vector_store.py | 179 ++---------------- .../tests/test_registry_evidence_config.py | 36 +++- harness/tht/adapters/factory.py | 10 +- harness/tht/adapters/vector/qdrant.py | 84 +++++--- harness/tht/cli/vector_cmd.py | 8 +- harness/tht/config.py | 44 +++-- harness/tht/ports/vector.py | 24 +++ 11 files changed, 410 insertions(+), 227 deletions(-) create mode 100644 harness/tests/qdrant_test_helpers.py diff --git a/backend/test/workspace-runtime-renderer.test.ts b/backend/test/workspace-runtime-renderer.test.ts index 06608e45..24e33ee0 100644 --- a/backend/test/workspace-runtime-renderer.test.ts +++ b/backend/test/workspace-runtime-renderer.test.ts @@ -1,4 +1,3 @@ -import { execFileSync } from "node:child_process"; import { mkdtempSync, rmSync, writeFileSync } from "node:fs"; import { tmpdir } from "node:os"; import { join } from "node:path"; @@ -109,7 +108,7 @@ test("keeps create_if_missing for non-registry runtime renders", () => { }); test.each(["session", "maintenance"])( - "renders require_existing vectors for %s acquisition", + "pure renderer emits require_existing vectors for %s acquisition", (mode) => { const rendered = parse(renderRuntimeConfig(workspaceV3, directBindings, paths, { workspaceId: "psd-clinical", workspaceRevision: "a".repeat(40), @@ -124,36 +123,12 @@ test.each(["session", "maintenance"])( }, ); -test("session and maintenance renders are byte-identical and bind identically in harness", () => { +test("pure session and maintenance renders are byte-identical", () => { const context = { workspaceId: "psd-clinical", workspaceRevision: "a".repeat(40) }; - const passwordFile = evidenceSecretFile("dwh-password", "not-a-canary"); - const bindings = { - ...directBindings, - dwh: { ...directBindings.dwh, values: { - ...directBindings.dwh.values, - THT_WS_PSD_CLINICAL_DWH_PASSWORD_FILE: passwordFile, - } }, - }; - const session = renderRuntimeConfig(workspaceV3, bindings, paths, context, {}, semanticRuntime); - const maintenance = renderRuntimeConfig(workspaceV3, bindings, paths, context, {}, semanticRuntime); + const session = renderRuntimeConfig(workspaceV3, directBindings, paths, context, {}, semanticRuntime); + const maintenance = renderRuntimeConfig(workspaceV3, directBindings, paths, context, {}, semanticRuntime); expect(maintenance).toBe(session); - - const script = ` -import json, sys, tempfile -from pathlib import Path -from tht.config import load_config -from tht.jobs.dwh_pipeline import config_dwh_binding -with tempfile.TemporaryDirectory() as root: - path = Path(root) / "runtime.yaml" - path.write_text(sys.stdin.read()) - print(json.dumps(config_dwh_binding(load_config(path)), sort_keys=True)) -`; - const python = join(process.cwd(), "../harness/.venv/bin/python"); - const run = (yaml: string) => execFileSync(python, ["-c", script], { - cwd: join(process.cwd(), "../harness"), input: yaml, encoding: "utf8", - }).trim(); - expect(run(session)).toBe(run(maintenance)); }); test("renders schema-v3 DWH REST without exposing secret contents", () => { diff --git a/harness/tests/qdrant_test_helpers.py b/harness/tests/qdrant_test_helpers.py new file mode 100644 index 00000000..69cdd672 --- /dev/null +++ b/harness/tests/qdrant_test_helpers.py @@ -0,0 +1,163 @@ +"""Shared fake Qdrant HTTP boundary for adapter and CLI tests.""" + +import json + +from tht.ports.vector import 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.payload_index_types: dict[str, str] = {} + 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 + self.scroll_pages: list[dict] | None = None + + 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": self.payload_index_types.get(field, "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": + if self.collection is None: + return FakeResponse(404, {"status": "error"}) + 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"}}) + if self.scroll_pages is not None: + offset = json.get("offset") + for page in self.scroll_pages: + if page["offset"] == offset: + filtered = _match_points(page["points"], json["filter"]) + return FakeResponse(200, { + "result": { + "points": filtered, + "next_page_offset": page["next_page_offset"], + } + }) + raise AssertionError(("unexpected offset", offset, self.scroll_pages)) + wanted = sorted( + _match_points(self.points.values(), json["filter"]), + key=lambda point: point["payload"]["record_key"], + ) + return FakeResponse(200, {"result": {"points": wanted, "next_page_offset": None}}) + + 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, + ) + + diff --git a/harness/tests/test_adapter_factory.py b/harness/tests/test_adapter_factory.py index d39b037a..de83ef74 100644 --- a/harness/tests/test_adapter_factory.py +++ b/harness/tests/test_adapter_factory.py @@ -67,6 +67,15 @@ def test_factory_selects_dwh_adapter(dwh_type, adapter_type): assert isinstance(build_dwh(_config(dwh_type=dwh_type)), adapter_type) +def test_factory_rejects_require_existing_without_embedding_dimension(): + config = _config() + config.vectors.collection_lifecycle = "require_existing" + config.embeddings = None + + with pytest.raises(ConfigError, match="explicit positive embedding dimension"): + build_vector_store(config) + + def test_factory_selects_qdrant_for_schema_v3_runtime(): config = _config(dwh_type="postgres_direct") diff --git a/harness/tests/test_qdrant_cli_commands.py b/harness/tests/test_qdrant_cli_commands.py index 0e17bccb..0b741fae 100644 --- a/harness/tests/test_qdrant_cli_commands.py +++ b/harness/tests/test_qdrant_cli_commands.py @@ -218,6 +218,53 @@ def test_vector_index_schema_json_is_single_document(monkeypatch, tmp_path): assert payload["counts"]["added"] == 2 +def test_vector_index_schema_json_maps_require_existing_delete_race(monkeypatch, tmp_path): + cfg = _qdrant_runtime_config(tmp_path) + _write_schema_artifacts(tmp_path) + monkeypatch.setattr("tht.cli.vector_cmd.make_embedder", lambda _: SimpleNamespace(embed_documents=lambda docs: [[0.1] * 1024 for _ in docs])) + keyword_indexes = { + "content_hash", "document_id", "kind", "record_key", "record_kind", + "vector_generation", "workspace_id", "workspace_revision", + } + deleted = False + + class Response: + def __init__(self, status_code, payload=None): + self.status_code = status_code + self.ok = status_code < 400 + self._payload = payload + self.text = "" if payload is None else "{}" + + def json(self): + return self._payload + + def request(method, url, **kwargs): + nonlocal deleted + if method == "POST" and url.endswith("/points/scroll"): + return Response(200, {"result": {"points": [], "next_page_offset": None}}) + if method == "GET" and url.endswith("/collections/psd-clinical"): + if deleted: + return Response(404, {"status": {"error": "missing"}}) + response = Response(200, {"result": { + "config": {"params": {"vectors": {"size": 1024, "distance": "Cosine"}}}, + "payload_schema": { + key: {"data_type": "keyword"} for key in keyword_indexes + }, + }}) + deleted = True + return response + if method == "PUT" and "/points?wait=true" in url: + return Response(404, {"status": {"error": "missing"}}) + raise AssertionError((method, url)) + + monkeypatch.setattr("requests.request", request) + response = CliRunner().invoke(app, ["vector", "index-schema", "--json", "-c", str(cfg)]) + + assert response.exit_code == 1 + assert response.stdout == '{"code":"semantic_index_incompatible","status":"failed"}\n' + assert response.stderr == "" + + def test_vector_index_schema_json_failure_is_safe(monkeypatch, tmp_path): import json diff --git a/harness/tests/test_qdrant_vector_store.py b/harness/tests/test_qdrant_vector_store.py index 676897e3..69d03f6a 100644 --- a/harness/tests/test_qdrant_vector_store.py +++ b/harness/tests/test_qdrant_vector_store.py @@ -1,167 +1,11 @@ -import json from uuid import NAMESPACE_URL, uuid5 import pytest import requests +from qdrant_test_helpers import FakeQdrantHttp, _write_record 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.payload_index_types: dict[str, str] = {} - 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 - self.scroll_pages: list[dict] | None = None - - 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": self.payload_index_types.get(field, "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": - if self.collection is None: - return FakeResponse(404, {"status": "error"}) - 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"}}) - if self.scroll_pages is not None: - offset = json.get("offset") - for page in self.scroll_pages: - if page["offset"] == offset: - filtered = _match_points(page["points"], json["filter"]) - return FakeResponse(200, { - "result": { - "points": filtered, - "next_page_offset": page["next_page_offset"], - } - }) - raise AssertionError(("unexpected offset", offset, self.scroll_pages)) - wanted = sorted( - _match_points(self.points.values(), json["filter"]), - key=lambda point: point["payload"]["record_key"], - ) - return FakeResponse(200, {"result": {"points": wanted, "next_page_offset": None}}) - - 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, - ) +from tht.ports.vector import VectorStoreError def _store(fake: FakeQdrantHttp, *, collection_lifecycle="create_if_missing") -> QdrantVectorStore: @@ -183,6 +27,20 @@ def test_point_id_is_deterministic_uuidv5(): +def test_require_existing_requires_an_explicit_embedding_dimension(): + fake = FakeQdrantHttp() + with pytest.raises(ValueError, match="expected dimension"): + QdrantVectorStore( + base_url="http://qdrant:6333", + collection="workspace-semantic", + workspace_id="demo", + expected_dimension=None, + collection_lifecycle="require_existing", + request=fake.request, + ) + assert fake.calls == [] + + def test_require_existing_refuses_missing_collection_without_mutations(): fake = FakeQdrantHttp() store = _store(fake, collection_lifecycle="require_existing") @@ -255,8 +113,11 @@ def test_require_existing_write_fails_after_collection_is_deleted_without_recrea collection_lifecycle="require_existing", request=request, ) - with pytest.raises(VectorStoreError): + from tht.ports.vector import SemanticIndexIncompatibleError + + with pytest.raises(SemanticIndexIncompatibleError) as caught: store.upsert("memory", [_write_record("memory:1", "memory")]) + assert caught.value.code == "semantic_index_incompatible" assert not [call for call in fake.calls if call[0] == "PUT" and call[1].endswith("/collections/workspace-semantic")] diff --git a/harness/tests/test_registry_evidence_config.py b/harness/tests/test_registry_evidence_config.py index e70ac8fa..e4e07438 100644 --- a/harness/tests/test_registry_evidence_config.py +++ b/harness/tests/test_registry_evidence_config.py @@ -460,7 +460,7 @@ def test_registry_rendered_session_and_maintenance_configs_require_existing_qdra assert yaml.safe_load(maintenance_path.read_text())["vectors"]["collection_lifecycle"] == "require_existing" assert session_path.read_bytes() == maintenance_path.read_bytes() - from test_qdrant_vector_store import FakeQdrantHttp, _write_record + from qdrant_test_helpers import FakeQdrantHttp, _write_record for cfg in (session_cfg, maintenance_cfg): fake = FakeQdrantHttp(dimension=384) if collection_state == "incompatible" else FakeQdrantHttp() @@ -481,6 +481,40 @@ def test_registry_rendered_session_and_maintenance_configs_bind_equally(tmp_path ) +def test_loader_rejects_split_brain_qdrant_compatibility_resources(tmp_path): + values = _qdrant_config_yaml(tmp_path, registry=True) + values["vectors"]["collection"] = "workspace-a" + values["resources"] = { + "vector": { + "engine": "qdrant", + "base_url": "http://qdrant:6333/", + "collection": "workspace-b", + } + } + path = tmp_path / "split-brain.yaml" + path.write_text(yaml.safe_dump(values)) + + with pytest.raises(ConfigError, match="vectors.*resources.vector|disagree"): + load_config(path) + + +def test_loader_rejects_lifecycle_policy_in_compatibility_resource(tmp_path): + values = _qdrant_config_yaml(tmp_path, registry=True) + values["resources"] = { + "vector": { + "engine": "qdrant", + "base_url": "http://qdrant:6333", + "collection": "workspace-semantic", + "collection_lifecycle": "require_existing", + } + } + path = tmp_path / "lifecycle.yaml" + path.write_text(yaml.safe_dump(values)) + + with pytest.raises(ConfigError, match="collection_lifecycle"): + load_config(path) + + def test_legacy_runtime_config_keeps_create_capable_qdrant_default(tmp_path): path = tmp_path / "legacy.yaml" path.write_text(yaml.safe_dump(_qdrant_config_yaml(tmp_path, registry=False))) diff --git a/harness/tht/adapters/factory.py b/harness/tht/adapters/factory.py index a9f52936..1d94cc11 100644 --- a/harness/tht/adapters/factory.py +++ b/harness/tht/adapters/factory.py @@ -32,12 +32,20 @@ def build_vector_store(cfg: Config, *, require_write: bool = False) -> VectorSto match resource.type: case "qdrant": + expected_dimension = cfg.embeddings.dim if cfg.embeddings is not None else None + if resource.collection_lifecycle == "require_existing" and ( + expected_dimension is None or expected_dimension <= 0 + ): + raise ConfigError( + "require_existing Qdrant vectors require an explicit positive embedding dimension" + ) return QdrantVectorStore( base_url=resource.base_url, collection=resource.collection, workspace_id=cfg._workspace_id, workspace_revision=cfg._workspace_revision, - expected_dimension=cfg.embeddings.dim if cfg.embeddings is not None else None, + expected_dimension=expected_dimension, + expected_distance="Cosine", collection_lifecycle=resource.collection_lifecycle, ) case other: # pragma: no cover - Pydantic's discriminator rejects this first. diff --git a/harness/tht/adapters/vector/qdrant.py b/harness/tht/adapters/vector/qdrant.py index f99359d9..06066d13 100644 --- a/harness/tht/adapters/vector/qdrant.py +++ b/harness/tht/adapters/vector/qdrant.py @@ -13,9 +13,12 @@ from tht.adapters.vector._shared import ( validate_known_kinds, ) from tht.ports.vector import ( + SemanticIndexIncompatibleError, VectorCapabilities, VectorHealth, + VectorResponseError, VectorStoreError, + VectorTransportError, VectorWriteRecord, require_positive_limit, ) @@ -55,6 +58,7 @@ class QdrantVectorStore: workspace_id: str, workspace_revision: str | None = None, expected_dimension: int | None = None, + expected_distance: str | None = "Cosine", collection_lifecycle: str = "create_if_missing", request: Callable[..., object] | None = None, connect_timeout: float = 2.0, @@ -65,8 +69,16 @@ class QdrantVectorStore: self._workspace_id = workspace_id self._workspace_revision = workspace_revision self._expected_dimension = expected_dimension + self._expected_distance = expected_distance if collection_lifecycle not in ("create_if_missing", "require_existing"): raise ValueError("Unsupported Qdrant collection lifecycle") + if collection_lifecycle == "require_existing" and ( + expected_dimension is None or expected_dimension <= 0 or expected_distance is None + ): + raise ValueError( + "require_existing requires an explicit positive expected dimension " + "and expected distance" + ) self._collection_lifecycle = collection_lifecycle self._request = request or requests.request self._timeout = (connect_timeout, read_timeout) @@ -100,9 +112,17 @@ class QdrantVectorStore: dimension = info["config"]["params"]["vectors"]["size"] dimensions = (dimension,) - compatible = ( + dimension_compatible = ( None if self._expected_dimension is None else dimensions == (self._expected_dimension,) ) + observed_distance = info["config"]["params"]["vectors"].get("distance") + distance_compatible = ( + None if self._expected_distance is None else observed_distance == self._expected_distance + ) + compatible = ( + None if dimension_compatible is None and distance_compatible is None + else dimension_compatible is not False and distance_compatible is not False + ) return VectorHealth( ok=compatible is not False, read_configured=True, @@ -161,7 +181,7 @@ class QdrantVectorStore: ) points = response.get("result", {}).get("points") if not isinstance(points, list): - raise VectorStoreError("Qdrant returned malformed query response") + raise VectorResponseError("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] @@ -179,11 +199,11 @@ class QdrantVectorStore: for point in points: payload = point.get("payload") if not isinstance(payload, dict): - raise VectorStoreError("Qdrant returned malformed scroll response") + raise VectorResponseError("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") + raise VectorResponseError("Qdrant returned malformed scroll response") hashes[record_key] = content_hash return hashes @@ -207,11 +227,18 @@ class QdrantVectorStore: ), } ) - self._call( - "PUT", - f"/collections/{self._collection}/points?wait=true", - {"points": points}, - ) + try: + self._call( + "PUT", + f"/collections/{self._collection}/points?wait=true", + {"points": points}, + ) + except VectorTransportError as exc: + if self._collection_lifecycle == "require_existing" and exc.status_code == 404: + raise SemanticIndexIncompatibleError( + "Qdrant collection disappeared during semantic index write" + ) from exc + raise return len(records) def delete_kinds(self, collection: str, kinds: list[str]) -> int: @@ -308,10 +335,10 @@ class QdrantVectorStore: 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 self._collection_lifecycle == "require_existing": + raise SemanticIndexIncompatibleError("Qdrant collection is missing") if not strict: raise VectorStoreError("Qdrant collection is missing") - if self._collection_lifecycle == "require_existing": - raise VectorStoreError("Qdrant collection configuration mismatch (semantic_index_incompatible)") self._call( "PUT", f"/collections/{self._collection}", @@ -327,24 +354,31 @@ class QdrantVectorStore: 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") + raise VectorResponseError("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") + self._expected_dimension is not None and size != self._expected_dimension + ) or ( + self._expected_distance is not None and distance != self._expected_distance ): - raise VectorStoreError("Qdrant collection configuration mismatch (semantic_index_incompatible)") + if self._collection_lifecycle == "require_existing": + raise SemanticIndexIncompatibleError( + "Qdrant collection dimension or distance is incompatible" + ) + raise VectorStoreError("Qdrant collection configuration mismatch") payload_schema = result.get("payload_schema") if not isinstance(payload_schema, dict): - raise VectorStoreError("Qdrant returned malformed collection response") + raise VectorResponseError("Qdrant returned malformed collection response") for field_name in _KEYWORD_INDEXES: field = payload_schema.get(field_name) if not isinstance(field, dict) or field.get("data_type") != "keyword": + if self._collection_lifecycle == "require_existing": + raise SemanticIndexIncompatibleError( + "Qdrant collection payload indexes are incompatible" + ) if not strict: raise VectorStoreError("Qdrant collection payload indexes mismatch") - if self._collection_lifecycle == "require_existing": - raise VectorStoreError("Qdrant collection configuration mismatch (semantic_index_incompatible)") self._call( "PUT", f"/collections/{self._collection}/index", @@ -370,13 +404,13 @@ class QdrantVectorStore: result = response.get("result", {}) page = result.get("points") if not isinstance(page, list): - raise VectorStoreError("Qdrant returned malformed scroll response") + raise VectorResponseError("Qdrant returned malformed scroll response") points.extend(page) next_page_offset = result.get("next_page_offset") if next_page_offset is None: return points if next_page_offset in seen_offsets: - raise VectorStoreError("Qdrant returned malformed scroll response") + raise VectorResponseError("Qdrant returned malformed scroll response") seen_offsets.add(next_page_offset) offset = next_page_offset @@ -384,7 +418,7 @@ class QdrantVectorStore: 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") + raise VectorResponseError("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: @@ -396,19 +430,19 @@ class QdrantVectorStore: timeout=self._timeout, ) except requests.RequestException as exc: - raise VectorStoreError(_sanitize_exception(exc)) from exc + raise VectorTransportError(_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}") + raise VectorTransportError(f"Qdrant request failed: HTTP {response.status_code}", status_code=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 + raise VectorResponseError("Qdrant returned malformed JSON response") from exc if not isinstance(data, dict): - raise VectorStoreError("Qdrant returned malformed JSON response") + raise VectorResponseError("Qdrant returned malformed JSON response") return data diff --git a/harness/tht/cli/vector_cmd.py b/harness/tht/cli/vector_cmd.py index 4e02b974..644db2bb 100644 --- a/harness/tht/cli/vector_cmd.py +++ b/harness/tht/cli/vector_cmd.py @@ -18,7 +18,7 @@ from tht.cli.schema_cmd import ( physical_path, ) from tht.config import Config, ConfigError -from tht.ports.vector import VectorWriteRecord +from tht.ports.vector import SemanticIndexIncompatibleError, VectorWriteRecord from tht.vectorstore.store import SyncStats, content_hash vector_app = typer.Typer(help="Indice semantico Qdrant (derivato, rigenerabile)") @@ -265,6 +265,12 @@ def index_schema_cmd( _vector_cfg_or_error(cfg) physical, annotations = _load_schema_artifacts(cfg) payload = index_schema_data(cfg, physical=physical, annotations=annotations) + except SemanticIndexIncompatibleError: + if json_output: + typer.echo(json.dumps({"status": "failed", "code": "semantic_index_incompatible"}, sort_keys=True, separators=(",", ":"))) + raise typer.Exit(code=1) from None + typer.secho("ERRORE: indice semantico incompatibile.", fg=typer.colors.RED, err=True) + raise typer.Exit(code=1) from None except _MachineVectorError as error: if json_output: typer.echo(json.dumps({"status": "failed", "code": error.code}, sort_keys=True, separators=(",", ":"))) diff --git a/harness/tht/config.py b/harness/tht/config.py index b0debe03..824c491a 100644 --- a/harness/tht/config.py +++ b/harness/tht/config.py @@ -550,17 +550,8 @@ def load_config(path: Path) -> Config: _validate_internal_embedding_contract(expanded, path) _validate_internal_vector_contract(expanded, path) translated, used_legacy = translate_legacy_config(expanded) + _validate_vector_resource_consistency(expanded, translated, path) vectors = translated.get("vectors") - resource_vector = expanded.get("resources", {}).get("vector") if isinstance( - expanded.get("resources"), dict - ) else None - if ( - isinstance(vectors, dict) - and vectors.get("type") == "qdrant" - and isinstance(resource_vector, dict) - and "collection_lifecycle" in resource_vector - ): - vectors["collection_lifecycle"] = resource_vector["collection_lifecycle"] # runtime_identity is the registry marker. The lifecycle is an internal # runtime policy, never a descriptor-controlled option. if ( @@ -673,6 +664,37 @@ def _validate_internal_embedding_contract(raw: dict[str, Any], path: Path) -> No ) +def _normalize_qdrant_base_url(value: str) -> str: + parsed = urlparse(value) + hostname = (parsed.hostname or "").lower() + try: + port = parsed.port + except ValueError: + port = None + host = f"[{hostname}]" if ":" in hostname and not hostname.startswith("[") else hostname + netloc = f"{host}:{port}" if port is not None else host + return f"{parsed.scheme.lower()}://{netloc}{parsed.path.rstrip('/') or '/'}" + + +def _validate_vector_resource_consistency(raw: dict[str, Any], translated: dict[str, Any], path: Path) -> None: + """Reject divergent top-level and compatibility Qdrant resource views.""" + resources = raw.get("resources") + resource = resources.get("vector") if isinstance(resources, dict) else None + vectors = translated.get("vectors") + if not isinstance(resource, dict) or not isinstance(vectors, dict): + return + if vectors.get("type") != "qdrant" or resource.get("engine") != "qdrant": + raise ConfigError( + f"Configurazione non valida in {path}: vectors and resources.vector must both describe qdrant" + ) + if _normalize_qdrant_base_url(vectors.get("base_url", "")) != _normalize_qdrant_base_url( + resource.get("base_url", "") + ) or vectors.get("collection") != resource.get("collection"): + raise ConfigError( + f"Configurazione non valida in {path}: vectors and resources.vector disagree" + ) + + def _validate_internal_vector_contract(raw: dict[str, Any], path: Path) -> None: resources = raw.get("resources") if not isinstance(resources, dict): @@ -684,7 +706,7 @@ def _validate_internal_vector_contract(raw: dict[str, Any], path: Path) -> None: engine = vector.get("engine") base_url = vector.get("base_url") collection = vector.get("collection") - allowed = {"engine", "base_url", "collection", "collection_lifecycle"} + allowed = {"engine", "base_url", "collection"} unexpected = sorted(set(vector) - allowed) if unexpected: raise ConfigError( diff --git a/harness/tht/ports/vector.py b/harness/tht/ports/vector.py index e2bdef21..0eba1489 100644 --- a/harness/tht/ports/vector.py +++ b/harness/tht/ports/vector.py @@ -45,6 +45,27 @@ class VectorStoreError(Exception): """Base error exposed by vector adapters.""" +class SemanticIndexIncompatibleError(VectorStoreError): + """The configured semantic collection cannot safely serve this workspace.""" + + code = "semantic_index_incompatible" + + def __init__(self, message: str = "Qdrant semantic index is incompatible"): + super().__init__(f"{self.code}: {message}") + + +class VectorTransportError(VectorStoreError): + """Qdrant could not be reached or returned an HTTP failure.""" + + def __init__(self, message: str, *, status_code: int | None = None): + super().__init__(message) + self.status_code = status_code + + +class VectorResponseError(VectorStoreError): + """Qdrant returned a malformed response.""" + + class VectorWriteUnavailable(VectorStoreError): """Raised when a deployment has no vector writer credential.""" @@ -86,13 +107,16 @@ class VectorStore(Protocol): __all__ = [ + "SemanticIndexIncompatibleError", "VectorCapabilities", "VectorHealth", "VectorHit", "VectorReadUnavailable", "VectorRecord", + "VectorResponseError", "VectorStore", "VectorStoreError", + "VectorTransportError", "VectorWriteRecord", "VectorWriteUnavailable", "require_positive_limit",