fix: close task3 qdrant and config validation gaps

harness/tht/adapters/vector/qdrant.py

@@ -188,6 +188,11 @@ class QdrantVectorStore:
     def existing_hashes(self, collection: str, kinds: list[str]) -> dict[str, str]:
         validate_collection(collection)
         validate_collection_kinds(collection, kinds)
+        if self._collection_lifecycle == "require_existing":
+            # Reconcile only after proving the exact existing collection contract.
+            # This keeps an initial 404 typed and prevents an incompatible empty
+            # collection from appearing healthy merely because there are no records.
+            self._ensure_collection(strict=False)
         points = self._scroll(
             [
                 *self._workspace_filter(),
@@ -209,7 +214,6 @@ class QdrantVectorStore:
 
     def upsert(self, collection: str, records: list[VectorWriteRecord]) -> int:
         validate_collection(collection)
-        self._ensure_collection(strict=True)
         points = []
         for write_record in records:
             validate_collection_kinds(collection, [write_record.record.kind])
@@ -228,6 +232,9 @@ class QdrantVectorStore:
                 }
             )
         try:
+            # Keep the preflight immediately adjacent to the mutation: a registry
+            # collection may disappear after reconciliation has read its hashes.
+            self._ensure_collection(strict=True)
             self._call(
                 "PUT",
                 f"/collections/{self._collection}/points?wait=true",
@@ -249,12 +256,23 @@ class QdrantVectorStore:
             self._semantic_kind_filter(kinds),
             {"key": "record_kind", "match": {"any": sorted(kinds)}},
         ]
+        if self._collection_lifecycle == "require_existing":
+            self._ensure_collection(strict=False)
         before = len(self._scroll(must))
-        self._call(
-            "POST",
-            f"/collections/{self._collection}/points/delete?wait=true",
-            {"filter": {"must": must}},
-        )
+        if self._collection_lifecycle == "require_existing":
+            self._ensure_collection(strict=False)
+        try:
+            self._call(
+                "POST",
+                f"/collections/{self._collection}/points/delete?wait=true",
+                {"filter": {"must": must}},
+            )
+        except VectorTransportError as exc:
+            if self._collection_lifecycle == "require_existing" and exc.status_code == 404:
+                raise SemanticIndexIncompatibleError(
+                    "Qdrant collection disappeared during semantic index deletion"
+                ) from exc
+            raise
         return before
 
     def delete_generation(self, collection: str, generation: str, workspace_id: str) -> int:
@@ -269,14 +287,25 @@ class QdrantVectorStore:
             {"key": "record_kind", "match": {"any": ["evidence"]}},
             {"key": "vector_generation", "match": {"value": generation}},
         ]
+        if self._collection_lifecycle == "require_existing":
+            self._ensure_collection(strict=False)
         before = len(
             self._scroll(must)
         )
-        self._call(
-            "POST",
-            f"/collections/{self._collection}/points/delete?wait=true",
-            {"filter": {"must": must}},
-        )
+        if self._collection_lifecycle == "require_existing":
+            self._ensure_collection(strict=False)
+        try:
+            self._call(
+                "POST",
+                f"/collections/{self._collection}/points/delete?wait=true",
+                {"filter": {"must": must}},
+            )
+        except VectorTransportError as exc:
+            if self._collection_lifecycle == "require_existing" and exc.status_code == 404:
+                raise SemanticIndexIncompatibleError(
+                    "Qdrant collection disappeared during semantic index deletion"
+                ) from exc
+            raise
         return before
 
     def list_evidence_generations(self, collection: str, workspace_id: str) -> list[str]:

harness/tht/config.py

@@ -664,13 +664,16 @@ 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()
+def _normalize_qdrant_base_url(value: Any) -> str | None:
+    """Normalize a URL, returning ``None`` for untrusted raw YAML values."""
+    if not isinstance(value, str):
+        return None
     try:
+        parsed = urlparse(value)
+        hostname = (parsed.hostname or "").lower()
         port = parsed.port
-    except ValueError:
-        port = None
+    except (TypeError, ValueError):
+        return 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 '/'}"
@@ -687,9 +690,18 @@ def _validate_vector_resource_consistency(raw: dict[str, Any], translated: dict[
         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"):
+    vector_url = _normalize_qdrant_base_url(vectors.get("base_url"))
+    resource_url = _normalize_qdrant_base_url(resource.get("base_url"))
+    vector_collection = vectors.get("collection")
+    resource_collection = resource.get("collection")
+    if (
+        vector_url is None
+        or resource_url is None
+        or not isinstance(vector_collection, str)
+        or not isinstance(resource_collection, str)
+        or vector_url != resource_url
+        or vector_collection != resource_collection
+    ):
         raise ConfigError(
             f"Configurazione non valida in {path}: vectors and resources.vector disagree"
         )