refactor(vector): define store contract

harness/tht/adapters/vector/__init__.py

@@ -0,0 +1,6 @@
+"""Vector-store adapter implementations."""
+
+from tht.adapters.vector.legacy_direct import LegacyDirectVectorStore
+from tht.adapters.vector.thoth_http import ThothHttpVectorStore
+
+__all__ = ["LegacyDirectVectorStore", "ThothHttpVectorStore"]

harness/tht/adapters/vector/legacy_direct.py

@@ -0,0 +1,47 @@
+"""Compatibility adapter for the existing direct PostgreSQL vector reader."""
+
+from sqlalchemy import Engine
+
+from tht.ports.vector import VectorCapabilities, VectorHealth, VectorRecord, VectorWriteUnavailable
+from tht.vectorstore.store import VectorHit, VectorStore as TableVectorStore
+
+
+class LegacyDirectVectorStore:
+    """Read-only port wrapper around the legacy table-scoped pgvector store."""
+
+    capabilities = VectorCapabilities(search=True, existing_hashes=False, upsert=False)
+
+    def __init__(self, engine: Engine, schema: str = "vectors", dim: int = 768):
+        self._engine = engine
+        self._schema = schema
+        self._dim = dim
+
+    def health(self) -> VectorHealth:
+        try:
+            with self._engine.connect() as connection:
+                connection.exec_driver_sql("SELECT 1")
+        except Exception as exc:
+            return VectorHealth(ok=False, detail=str(exc))
+        return VectorHealth(ok=True)
+
+    def search(
+        self,
+        collections: list[str],
+        embedding: list[float],
+        *,
+        limit: int,
+        kinds: list[str] | None = None,
+    ) -> list[VectorHit]:
+        hits: list[VectorHit] = []
+        for collection in collections:
+            table = TableVectorStore(
+                self._engine, schema=self._schema, table=collection, dim=self._dim
+            )
+            hits.extend(table.search(embedding, top_n=limit, kinds=kinds))
+        return sorted(hits, key=lambda hit: hit.similarity, reverse=True)[:limit]
+
+    def existing_hashes(self, collection: str, kinds: list[str]) -> dict[str, str]:
+        raise VectorWriteUnavailable("Legacy direct reader has no writer interface")
+
+    def upsert(self, collection: str, records: list[VectorRecord]) -> int:
+        raise VectorWriteUnavailable("Legacy direct reader has no writer interface")

harness/tht/adapters/vector/thoth_http.py

@@ -0,0 +1,93 @@
+"""Thoth vector HTTP adapter using distinct read and write clients."""
+
+from tht.ports.vector import (
+    VectorCapabilities,
+    VectorHealth,
+    VectorHit,
+    VectorRecord,
+    VectorStoreError,
+    VectorWriteUnavailable,
+)
+from tht.vectorstore.rest_client import VectorRestClient
+from tht.vectorstore.store import content_hash, hit_from_metadata
+
+
+def _merge(hits: list[VectorHit], limit: int) -> list[VectorHit]:
+    return sorted(hits, key=lambda hit: hit.similarity, reverse=True)[:limit]
+
+
+class ThothHttpVectorStore:
+    """Vector port backed by the existing allowlisted REST RPCs."""
+
+    def __init__(self, reader: VectorRestClient, writer: VectorRestClient | None):
+        self._reader = reader
+        self._writer = writer
+
+    @property
+    def capabilities(self) -> VectorCapabilities:
+        writable = self._writer is not None
+        return VectorCapabilities(search=True, existing_hashes=writable, upsert=writable)
+
+    def health(self) -> VectorHealth:
+        try:
+            self._reader.list_tables()
+        except Exception as exc:
+            return VectorHealth(ok=False, detail=str(exc))
+        return VectorHealth(ok=True)
+
+    def search(
+        self,
+        collections: list[str],
+        embedding: list[float],
+        *,
+        limit: int,
+        kinds: list[str] | None = None,
+    ) -> list[VectorHit]:
+        hits: list[VectorHit] = []
+        for collection in collections:
+            rows = self._reader.search_similar(collection, embedding, limit, kinds=kinds)
+            hits.extend(
+                hit_from_metadata(row.get("similarity", 0.0), row.get("metadata"))
+                for row in rows
+            )
+        if kinds:
+            allowed = set(kinds)
+            hits = [hit for hit in hits if hit.kind in allowed]
+        return _merge(hits, limit)
+
+    def _require_writer(self) -> VectorRestClient:
+        if self._writer is None:
+            raise VectorWriteUnavailable("Vector writer credential is not configured")
+        return self._writer
+
+    def existing_hashes(self, collection: str, kinds: list[str]) -> dict[str, str]:
+        return self._require_writer().existing_hashes(collection, kinds)
+
+    def upsert(self, collection: str, records: list[VectorRecord]) -> int:
+        writer = self._require_writer()
+        rows = [self._row(record) for record in records]
+        return writer.upsert_records(collection, rows)
+
+    @staticmethod
+    def _row(record: VectorRecord) -> dict:
+        extra = dict(record.metadata)
+        try:
+            embedding = extra.pop("embedding")
+        except KeyError as exc:
+            raise VectorStoreError(f"Vector record {record.id!r} has no embedding") from exc
+        digest = extra.pop("content_hash", content_hash(record.content))
+        metadata = {
+            "kind": record.kind,
+            "ref": record.ref,
+            "record_key": record.id,
+            "title": record.title,
+            "content": record.content,
+            **extra,
+        }
+        return {
+            "record_key": record.id,
+            "kind": record.kind,
+            "content_hash": digest,
+            "metadata": metadata,
+            "embedding": embedding,
+        }