From 8826f8ac6bc122f2665c8e98c654c7b622e20172 Mon Sep 17 00:00:00 2001 From: mptyl Date: Sat, 8 Aug 2026 21:38:03 +0200 Subject: [PATCH] refactor: remove pgvector runtime --- .../task-12-report.md | 27 + harness/pyproject.toml | 2 +- .../l0/test_pgvector_corpus_lifecycle.py | 258 --------- harness/tests/l0/test_pgvector_store.py | 407 -------------- .../tests/l0/test_vector_adapter_parity.py | 294 ---------- harness/tests/l0/test_vector_migrations.py | 303 ---------- harness/tests/l2/test_memory_save_one_real.py | 53 -- harness/tests/test_adapter_factory.py | 149 +---- harness/tests/test_blocco6_robustness.py | 54 -- harness/tests/test_config_legacy_compat.py | 148 ----- harness/tests/test_search_similar_kinds.py | 123 ---- harness/tests/test_solved_search_cli.py | 7 +- harness/tests/test_vector_dual_key.py | 64 --- .../tests/test_vector_migration_packaging.py | 15 +- harness/tests/test_vector_port_contract.py | 271 ++------- harness/tht/adapters/factory.py | 63 +-- harness/tht/adapters/vector/__init__.py | 5 +- harness/tht/adapters/vector/_shared.py | 41 ++ harness/tht/adapters/vector/legacy_direct.py | 70 --- harness/tht/adapters/vector/pgvector.py | 524 ------------------ harness/tht/adapters/vector/qdrant.py | 24 +- harness/tht/adapters/vector/thoth_http.py | 191 ------- harness/tht/cli/evidence_cmd.py | 15 +- harness/tht/cli/memory_cmd.py | 25 +- harness/tht/cli/search_cmd.py | 5 +- harness/tht/cli/vector_cmd.py | 55 +- .../tht/migrations/vector/001_extensions.sql | 3 - .../migrations/vector/002_schema_tables.sql | 32 -- harness/tht/migrations/vector/003_roles.sql | 23 - .../vector/004_evidence_generation_gc.sql | 3 - harness/tht/solved.py | 2 +- harness/tht/vectorstore/reader.py | 49 +- harness/tht/vectorstore/rest_client.py | 173 ------ harness/tht/vectorstore/rest_writer.py | 84 --- 34 files changed, 200 insertions(+), 3362 deletions(-) create mode 100644 .superpowers/sdd/2026-08-08-internal-qdrant-ollama/task-12-report.md delete mode 100644 harness/tests/l0/test_pgvector_corpus_lifecycle.py delete mode 100644 harness/tests/l0/test_pgvector_store.py delete mode 100644 harness/tests/l0/test_vector_adapter_parity.py delete mode 100644 harness/tests/l0/test_vector_migrations.py delete mode 100644 harness/tests/l2/test_memory_save_one_real.py delete mode 100644 harness/tests/test_blocco6_robustness.py delete mode 100644 harness/tests/test_config_legacy_compat.py delete mode 100644 harness/tests/test_search_similar_kinds.py delete mode 100644 harness/tests/test_vector_dual_key.py create mode 100644 harness/tht/adapters/vector/_shared.py delete mode 100644 harness/tht/adapters/vector/legacy_direct.py delete mode 100644 harness/tht/adapters/vector/pgvector.py delete mode 100644 harness/tht/adapters/vector/thoth_http.py delete mode 100644 harness/tht/migrations/vector/001_extensions.sql delete mode 100644 harness/tht/migrations/vector/002_schema_tables.sql delete mode 100644 harness/tht/migrations/vector/003_roles.sql delete mode 100644 harness/tht/migrations/vector/004_evidence_generation_gc.sql delete mode 100644 harness/tht/vectorstore/rest_client.py delete mode 100644 harness/tht/vectorstore/rest_writer.py diff --git a/.superpowers/sdd/2026-08-08-internal-qdrant-ollama/task-12-report.md b/.superpowers/sdd/2026-08-08-internal-qdrant-ollama/task-12-report.md new file mode 100644 index 00000000..d269d921 --- /dev/null +++ b/.superpowers/sdd/2026-08-08-internal-qdrant-ollama/task-12-report.md @@ -0,0 +1,27 @@ +# Task 12 Report — Remove unreachable pgvector runtime code + +Status: completed + +Summary: +- Proved the retired pgvector runtime had no remaining operational adapter call sites after migration by re-running the required grep; only the packaging assertion still mentions `migrations/vector`. +- Removed the obsolete pgvector/HTTP/direct vector runtime modules, vector SQL migrations, and their affected runtime tests. +- Kept the operational semantic path on Qdrant and migrated the remaining runtime callers to that path. +- Kept `psycopg2-binary` because DWH direct PostgreSQL and session PostgreSQL code still depend on it. + +Implementation notes: +- Extracted shared collection/kind validation into `harness/tht/adapters/vector/_shared.py` so `QdrantVectorStore` no longer depends on the deleted pgvector module. +- Simplified `build_vector_store()` to return only `QdrantVectorStore`. +- Migrated vector/evidence/memory CLI paths away from legacy pgvector loaders and REST vector clients. +- Updated packaging coverage so the built wheel asserts session SQL migrations are present and vector SQL migrations are absent. + +Verification: +- `cd harness && .venv/bin/pytest tests/test_qdrant_vector_store.py tests/test_vector_port_contract.py tests/test_semantic_kind_isolation.py tests/test_vector_migration_packaging.py -q` +- `cd harness && .venv/bin/pytest tests/test_adapter_factory.py tests/test_solved_search_cli.py -q` +- `cd harness && .venv/bin/python -c "import tht.cli, tht.adapters.factory, tht.adapters.vector, tht.vectorstore.reader"` +- `cd harness && uv build` +- `harness/.venv/bin/ruff check harness/tests/test_adapter_factory.py harness/tests/test_solved_search_cli.py harness/tests/test_vector_migration_packaging.py harness/tests/test_vector_port_contract.py harness/tht/adapters/factory.py harness/tht/adapters/vector/__init__.py harness/tht/adapters/vector/_shared.py harness/tht/adapters/vector/qdrant.py harness/tht/cli/evidence_cmd.py harness/tht/cli/memory_cmd.py harness/tht/cli/search_cmd.py harness/tht/cli/vector_cmd.py harness/tht/solved.py harness/tht/vectorstore/reader.py` +- `git diff --check` + +Notes / concerns: +- Repository-wide `harness/.venv/bin/ruff check .` still reports many pre-existing findings outside this task’s touched files; it is not clean on this branch baseline. +- Some legacy config compatibility parsing still exists outside the deleted runtime path. This task removed the unreachable runtime/migration code without broad config-schema refactoring. diff --git a/harness/pyproject.toml b/harness/pyproject.toml index 2ca5241e..5bc78f98 100644 --- a/harness/pyproject.toml +++ b/harness/pyproject.toml @@ -34,7 +34,7 @@ dev = [ include = ["tht*"] [tool.setuptools.package-data] -tht = ["migrations/vector/*.sql", "migrations/sessions/*.sql"] +tht = ["migrations/sessions/*.sql"] [tool.ruff] line-length = 100 diff --git a/harness/tests/l0/test_pgvector_corpus_lifecycle.py b/harness/tests/l0/test_pgvector_corpus_lifecycle.py deleted file mode 100644 index 9a06bc29..00000000 --- a/harness/tests/l0/test_pgvector_corpus_lifecycle.py +++ /dev/null @@ -1,258 +0,0 @@ -"""L0 gate for the complete durable Evidence/pgvector lifecycle.""" - -import hashlib -import json - -import pytest -from sqlalchemy import create_engine, text -from testcontainers.postgres import PostgresContainer - -from tht.adapters.evidence import FilesystemEvidenceSource -from tht.adapters.vector.pgvector import PgVectorStore -from tht.cli.vector_migrate_cmd import migrate -from tht.config import DatabaseConfig -from tht.corpus.chunk import ChunkPolicy -from tht.corpus.pipeline import CorpusPipeline -from tht.corpus.store import CorpusStore -from tht.ports.vector import VectorRecord, VectorWriteRecord -from tht.search import combined_search -from tht.search.evidence import ActiveEvidenceSearcher, resolve_evidence_file - - -DIMENSIONS = 768 - - -class DeterministicEmbedder: - def embed_documents(self, texts): - return [self.embed_query(text) for text in texts] - - def embed_query(self, text): - vector = [0.0] * DIMENSIONS - vector[0] = 0.8 - vector[1] = 0.6 - return vector - - -class EvidenceDelegate: - """Adapt the real multi-collection port to the runtime search protocol.""" - - def __init__(self, store): - self.store = store - - def search(self, embedding, top_n=10, kinds=None, metadata_filter=None): - return self.store.search( - ["evidence"], embedding, limit=top_n, kinds=kinds, - metadata_filter=metadata_filter, - ) - - -class InterruptAfterRealPartialUpsert: - """Crash after a committed real row, as a process death would.""" - - def __init__(self, store): - self.store = store - self.interrupt = True - - def __getattr__(self, name): - return getattr(self.store, name) - - def upsert(self, collection, records): - if self.interrupt and len(records) > 1: - self.interrupt = False - self.store.upsert(collection, records[:1]) - raise KeyboardInterrupt("injected process death after committed vector row") - return self.store.upsert(collection, records) - - -@pytest.fixture(scope="module") -def persistent_pgvector(): - with PostgresContainer("pgvector/pgvector:pg16") as postgres: - migrate(postgres.get_connection_url()) - admin = create_engine(postgres.get_connection_url()) - with admin.begin() as connection: - connection.exec_driver_sql( - "ALTER ROLE vector_reader LOGIN PASSWORD 'reader-lifecycle'" - ) - connection.exec_driver_sql( - "ALTER ROLE vector_writer LOGIN PASSWORD 'writer-lifecycle'" - ) - url = admin.url - common = dict( - host=url.host, port=url.port, database=url.database, schema="vectors" - ) - reader = DatabaseConfig( - **common, user="vector_reader", password="reader-lifecycle" - ) - writer = DatabaseConfig( - **common, user="vector_writer", password="writer-lifecycle" - ) - yield postgres, admin, reader, writer - admin.dispose() - - -def _pipeline(root, source_root, vectors): - return CorpusPipeline( - store=CorpusStore(root / "corpus"), - sources=[FilesystemEvidenceSource(source_root)], - embedder=DeterministicEmbedder(), - vector_store=vectors, - embedding_model="deterministic-l0", - embedding_dimensions=DIMENSIONS, - chunk_policy=ChunkPolicy(version="lifecycle-v1", max_chars=48), - pipeline_version="evidence-v1", - retain_published_generations=2, - ) - - -def _publish(pipeline, root, serial): - return pipeline.run_as_job( - workspace_id="pgvector-lifecycle", - workspace_root=root, - config_fingerprint="sha256:" + "1" * 64, - input_fingerprint="sha256:" + f"{serial:x}" * 64, - ) - - -@pytest.mark.l0 -def test_real_pgvector_corpus_job_lifecycle(tmp_path, persistent_pgvector): - postgres, admin, reader_config, writer_config = persistent_pgvector - source_root = tmp_path / "sources" - source_root.mkdir() - kept = source_root / "kept.md" - stable = source_root / "stable.md" - stable.write_text("unchanged dependency evidence", encoding="utf-8") - removed = source_root / "removed.md" - removed.write_text("removed evidence generation zero", encoding="utf-8") - - vectors = PgVectorStore(reader_config, writer_config, expected_dimension=DIMENSIONS) - generations = [] - for serial in range(3): - kept.write_text(f"active evidence generation {serial}", encoding="utf-8") - result = _publish(_pipeline(tmp_path, source_root, vectors), tmp_path, serial + 1) - assert result.status == "succeeded" - generations.append(result.generation) - removed_document = next( - doc for doc in CorpusStore(tmp_path / "corpus").active_manifest().documents - if "removed.md" in doc.source_uri - ) - removed_document_id = removed_document.document_id - removed_ref = removed_document.document_id - - # A stale, closer row must not consume LIMIT before ACTIVE filtering. - stale_generation = generations[-2] - stale = VectorWriteRecord( - record=VectorRecord( - id="chunk:stale-perfect-match", kind="evidence", ref="doc:stale", - title="stale forbidden", content="stale forbidden", - metadata={ - "document_id": "doc:stale", - "vector_generation": stale_generation, - }, - ), - embedding=[1.0] + [0.0] * (DIMENSIONS - 1), - content_hash="sha256:" + "a" * 64, - ) - vectors.upsert("evidence", [stale]) - runtime = ActiveEvidenceSearcher(CorpusStore(tmp_path / "corpus"), EvidenceDelegate(vectors)) - query = DeterministicEmbedder().embed_query("active") - hits = runtime.search(query, top_n=1, kinds=["evidence"]) - assert len(hits) == 1 and hits[0].title != "stale forbidden" - packed = combined_search( - "active", lsh_hits=None, store=runtime, embedder=DeterministicEmbedder(), - top=1, rrf_k=60, kinds=["evidence"], query_vec=query, - ) - assert len(packed) == 1 and packed[0].label != "stale forbidden" - - # Fourth publication removes a document and creates multiple chunks for crash recovery. - removed.unlink() - kept.write_text("active fourth generation " * 8, encoding="utf-8") - crashing = InterruptAfterRealPartialUpsert(vectors) - candidate = _pipeline(tmp_path, source_root, crashing) - with pytest.raises(KeyboardInterrupt, match="injected process death"): - _publish(candidate, tmp_path, 4) - runs = tmp_path / ".tht-jobs" / "evidence" / "runs" - crashed_run = max(runs.iterdir(), key=lambda path: path.stat().st_mtime_ns).name - before = vectors.existing_hashes("evidence", ["evidence"]) - intent = json.loads( - (runs / crashed_run / "artifacts" / "vector-intent.json").read_text() - )["records"] - already_present = set(intent) & set(before) - assert len(already_present) == 1 - resumed = candidate.run_as_job( - workspace_id="pgvector-lifecycle", workspace_root=tmp_path, - config_fingerprint="sha256:" + "1" * 64, - input_fingerprint="sha256:" + "4" * 64, - resume_run_id=crashed_run, - ) - assert resumed.status == "succeeded" and resumed.resumed_from == crashed_run - generations.append(resumed.generation) - after = vectors.existing_hashes("evidence", ["evidence"]) - assert {key: after[key] for key in already_present} == { - key: before[key] for key in already_present - } - assert set(intent).issubset(after) - with admin.connect() as connection: - duplicate_count = connection.execute(text( - "SELECT count(*) - count(DISTINCT record_key) FROM vectors.evidence" - )).scalar_one() - assert duplicate_count == 0 - - runtime = ActiveEvidenceSearcher(CorpusStore(tmp_path / "corpus"), EvidenceDelegate(vectors)) - active_hits = runtime.search(query, top_n=20, kinds=["evidence"]) - assert active_hits - assert any("active fourth generation" in hit.content for hit in active_hits) - assert all(hit.ref not in {"doc:stale", removed_ref} for hit in active_hits) - assert all(hit.metadata.get("document_id") != removed_document_id for hit in active_hits) - active_pack = combined_search( - "active", lsh_hits=None, store=runtime, embedder=DeterministicEmbedder(), - top=20, rrf_k=60, kinds=["evidence"], query_vec=query, - ) - assert active_pack - assert any("active fourth generation" in result.content for result in active_pack) - assert all(removed_document.content not in result.content for result in active_pack) - assert any("unchanged dependency evidence" in result.content for result in active_pack) - manifest = CorpusStore(tmp_path / "corpus").active_manifest() - assert resolve_evidence_file( - CorpusStore(tmp_path / "corpus"), removed_document_id, - materialized_root=tmp_path / "session", - ) == "" - - active_document = manifest.documents[0] - owned = CorpusStore(tmp_path / "corpus").materialize_document( - active_document.document_id, tmp_path / "session" / "active-evidence.md" - ) - assert owned.read_bytes() == active_document.content.encode() - assert hashlib.sha256(owned.read_bytes()).hexdigest() == active_document.content_hash[7:] - - # Recreate engines and stores against the same persisted database. - vectors._reader.dispose() - vectors._writer.dispose() - recreated = PgVectorStore(reader_config, writer_config, expected_dimension=DIMENSIONS) - assert recreated.health().ok is True - recreated_hits = ActiveEvidenceSearcher( - CorpusStore(tmp_path / "corpus"), EvidenceDelegate(recreated) - ).search(query, top_n=2, kinds=["evidence"]) - active_dependencies = set(manifest.metadata["document_generations"].values()) - assert recreated_hits - assert all(hit.metadata["vector_generation"] in active_dependencies for hit in recreated_hits) - - orphan = "gen:" + "f" * 32 - recreated.upsert("evidence", [VectorWriteRecord( - record=VectorRecord( - id="chunk:exact-vector-orphan", kind="evidence", ref="doc:orphan", - title="orphan", content="orphan", - metadata={"document_id": "doc:orphan", "vector_generation": orphan, - "workspace_id": "pgvector-lifecycle"}, - ), - embedding=query, content_hash="sha256:" + "f" * 64, - )]) - final_pipeline = _pipeline(tmp_path, source_root, recreated) - report = final_pipeline.gc(workspace_root=tmp_path) - assert report["evicted"] == [orphan] - expected_fs = set(generations[-2:]) - assert set(CorpusStore(tmp_path / "corpus").list_generations()) == expected_fs - expected_vectors = expected_fs | {generations[0]} - assert set(recreated.list_evidence_generations( - "evidence", "pgvector-lifecycle" - )) == expected_vectors - assert final_pipeline.gc(workspace_root=tmp_path)["evicted"] == [] diff --git a/harness/tests/l0/test_pgvector_store.py b/harness/tests/l0/test_pgvector_store.py deleted file mode 100644 index 0f010804..00000000 --- a/harness/tests/l0/test_pgvector_store.py +++ /dev/null @@ -1,407 +0,0 @@ -import pytest -from psycopg2.errors import InsufficientPrivilege -from sqlalchemy import create_engine, text -from sqlalchemy.exc import ProgrammingError -from testcontainers.postgres import PostgresContainer - -from tht.adapters.vector.thoth_http import ThothHttpVectorStore -from tht.config import DatabaseConfig -from tht.ports.vector import ( - VectorReadUnavailable, - VectorRecord, - VectorStoreError, - VectorWriteRecord, - VectorWriteUnavailable, -) - - -def _record(content_hash: str, embedding: list[float], *, kind: str = "memory"): - return VectorWriteRecord( - record=VectorRecord( - id=f"record:{content_hash}", - kind=kind, - ref="session:test", - title=content_hash, - content=f"content {content_hash}", - metadata={"content_hash": content_hash}, - ), - embedding=embedding, - content_hash=content_hash, - ) - - -@pytest.fixture(scope="module") -def vector_configs(): - with PostgresContainer("pgvector/pgvector:pg16") as pg: - host = pg.get_container_host_ip() - port = int(pg.get_exposed_port(5432)) - admin_config = DatabaseConfig( - host=host, - port=port, - database=pg.dbname, - schema="vectors", - user=pg.username, - password=pg.password, - ) - engine = create_engine(pg.get_connection_url()) - with engine.begin() as connection: - connection.exec_driver_sql("CREATE SCHEMA vectors") - # Match the co-located Supabase deployment: tables are in vectors, extension in public. - connection.exec_driver_sql("CREATE EXTENSION vector WITH SCHEMA public") - for table in ("schema_records", "evidence", "memory"): - connection.exec_driver_sql(f""" - CREATE TABLE vectors.{table} ( - id bigserial PRIMARY KEY, - record_key text UNIQUE NOT NULL, - kind text NOT NULL, - content_hash text NOT NULL, - metadata jsonb NOT NULL, - embedding public.vector(2) NOT NULL, - indexed_at timestamptz NOT NULL DEFAULT now() - ) - """) - connection.exec_driver_sql("CREATE ROLE vector_l0_reader LOGIN PASSWORD 'reader'") - connection.exec_driver_sql("CREATE ROLE vector_l0_writer LOGIN PASSWORD 'writer'") - connection.exec_driver_sql( - "CREATE ROLE vector_l0_no_sequence LOGIN PASSWORD 'no_sequence'" - ) - connection.exec_driver_sql( - "GRANT USAGE ON SCHEMA vectors TO vector_l0_reader, vector_l0_writer, " - "vector_l0_no_sequence" - ) - connection.exec_driver_sql( - "GRANT SELECT ON ALL TABLES IN SCHEMA vectors TO vector_l0_reader" - ) - connection.exec_driver_sql( - "GRANT USAGE, SELECT ON ALL SEQUENCES IN SCHEMA vectors TO vector_l0_writer" - ) - for table in ("schema_records", "evidence", "memory"): - connection.exec_driver_sql( - f"GRANT INSERT, UPDATE ON vectors.{table} " - "TO vector_l0_writer, vector_l0_no_sequence" - ) - if table == "evidence": - connection.exec_driver_sql( - "GRANT DELETE ON vectors.evidence TO vector_l0_writer" - ) - connection.exec_driver_sql( - "GRANT SELECT (kind, metadata) ON vectors.evidence TO vector_l0_writer" - ) - connection.exec_driver_sql( - f"GRANT SELECT (record_key, kind, content_hash) " - f"ON vectors.{table} TO vector_l0_writer, vector_l0_no_sequence" - ) - engine.dispose() - reader_config = admin_config.model_copy( - update={"user": "vector_l0_reader", "password": "reader"} - ) - writer_config = admin_config.model_copy( - update={"user": "vector_l0_writer", "password": "writer"} - ) - no_sequence_config = admin_config.model_copy( - update={"user": "vector_l0_no_sequence", "password": "no_sequence"} - ) - yield admin_config, reader_config, writer_config, no_sequence_config - - -@pytest.fixture -def store(vector_configs): - from tht.adapters.vector.pgvector import PgVectorStore - - _, reader_config, writer_config, _ = vector_configs - store = PgVectorStore(reader_config, writer_config, expected_dimension=2) - store.upsert("memory", [_record("reset", [0.0, 1.0])]) - yield store - - -def test_pgvector_round_trip_hash_and_upsert(store): - assert store.upsert("memory", [_record("a", [1.0, 0.0])]) == 1 - assert store.existing_hashes("memory", ["memory"])["record:a"] == "a" - - hits = store.search(["memory"], [1.0, 0.0], limit=5, kinds=["memory"]) - assert hits[0].metadata["content_hash"] == "a" - assert hits[0].id == "record:a" - - assert store.upsert("memory", [_record("a", [0.8, 0.2])]) == 1 - assert store.search(["memory"], [0.8, 0.2], limit=1)[0].id == "record:a" - - -def test_pgvector_lists_and_deletes_exact_evidence_generation(store): - generation = "gen:" + "a" * 32 - value = VectorWriteRecord( - record=VectorRecord( - id="evidence-generation-a", kind="evidence", ref="doc:a", title="a", - content="content", metadata={"vector_generation": generation, "workspace_id": "default"}, - ), - embedding=[1.0, 0.0], content_hash="sha256:" + "a" * 64, - ) - store.upsert("evidence", [value]) - assert generation in store.list_evidence_generations("evidence", "default") - assert store.delete_generation("evidence", generation, "default") == 1 - assert generation not in store.list_evidence_generations("evidence", "default") - - -def test_pgvector_generation_cleanup_isolated_between_workspaces(store): - generation = "gen:" + "b" * 32 - records = [VectorWriteRecord( - record=VectorRecord( - id=f"evidence-{workspace}", kind="evidence", ref=f"doc:{workspace}", - title=workspace, content=workspace, - metadata={"vector_generation": generation, "workspace_id": workspace}, - ), embedding=[1.0, 0.0], content_hash="sha256:" + key * 64, - ) for workspace, key in (("workspace-a", "b"), ("workspace-b", "c"))] - store.upsert("evidence", records) - assert store.delete_generation("evidence", generation, "workspace-a") == 1 - assert generation not in store.list_evidence_generations("evidence", "workspace-a") - assert generation in store.list_evidence_generations("evidence", "workspace-b") - - -def test_pgvector_search_filters_kinds_before_limit(store): - store.upsert("memory", [_record("solved", [1.0, 0.0], kind="solved_question")]) - hits = store.search("memory".split(), [1.0, 0.0], limit=1, kinds=["memory"]) - assert len(hits) == 1 - assert hits[0].kind == "memory" - - -def test_pgvector_multi_collection_search_skips_collections_unrelated_to_kinds(store): - store.upsert("evidence", [_record("evidence", [1.0, 0.0], kind="evidence")]) - - hits = store.search(["evidence", "memory"], [1.0, 0.0], limit=3, kinds=["memory"]) - - assert hits - assert {hit.kind for hit in hits} == {"memory"} - - -def test_pgvector_multi_collection_kind_filter_matches_http_adapter(store): - class Reader: - def search_similar(self, collection, embedding, limit, kinds=None): - if collection != "memory" or "memory" not in (kinds or []): - return [] - return [ - { - "similarity": 1.0, - "metadata": { - "record_key": "record:a", - "kind": "memory", - "ref": "session:test", - "title": "a", - "content": "content a", - "content_hash": "a", - }, - } - ] - - direct = store.search(["evidence", "memory"], [1.0, 0.0], limit=1, kinds=["memory"]) - http = ThothHttpVectorStore(Reader(), None).search( - ["evidence", "memory"], [1.0, 0.0], limit=1, kinds=["memory"] - ) - assert [(hit.id, hit.kind) for hit in direct] == [(hit.id, hit.kind) for hit in http] - - -def test_pgvector_search_rejects_unknown_kind_globally(store): - with pytest.raises(VectorStoreError, match="Kind not allowed"): - store.search(["memory"], [1.0, 0.0], limit=1, kinds=["unknown"]) - - -@pytest.mark.parametrize("limit", [True, False, 1.0, 0, -1]) -def test_pgvector_search_requires_strict_positive_limit(store, limit): - with pytest.raises(ValueError, match="positive integer"): - store.search(["memory"], [1.0, 0.0], limit=limit) - - -def test_pgvector_allowlists_collections(store): - with pytest.raises(VectorStoreError, match="Collection not allowed"): - store.search(["memory; DROP SCHEMA vectors"], [1.0, 0.0], limit=1) - with pytest.raises(VectorStoreError, match="Collection not allowed"): - store.upsert("unknown", []) - - -def test_pgvector_rejects_kinds_not_belonging_to_collection(store): - with pytest.raises(VectorStoreError, match="Kind not allowed"): - store.existing_hashes("evidence", ["memory"]) - with pytest.raises(VectorStoreError, match="Kind not allowed"): - store.upsert("evidence", [_record("wrong", [1.0, 0.0])]) - - -def test_pgvector_separates_read_and_write_credentials(vector_configs): - from tht.adapters.vector.pgvector import PgVectorStore - - _, reader_config, writer_config, _ = vector_configs - reader = PgVectorStore(reader_config, expected_dimension=2) - assert reader.capabilities.search is True - assert reader.capabilities.upsert is False - with pytest.raises(VectorWriteUnavailable): - reader.upsert("memory", []) - - writer = PgVectorStore(None, writer_config, expected_dimension=2) - assert writer.capabilities.search is False - assert writer.capabilities.upsert is True - with pytest.raises(VectorReadUnavailable): - writer.search(["memory"], [1.0, 0.0], limit=1) - - -def test_pgvector_database_roles_are_least_privilege(vector_configs): - _, reader_config, writer_config, _ = vector_configs - reader_engine = create_engine( - f"postgresql+psycopg2://{reader_config.user}:{reader_config.password}" - f"@{reader_config.host}:{reader_config.port}/{reader_config.database}" - ) - writer_engine = create_engine( - f"postgresql+psycopg2://{writer_config.user}:{writer_config.password}" - f"@{writer_config.host}:{writer_config.port}/{writer_config.database}" - ) - with pytest.raises(ProgrammingError): - with reader_engine.begin() as connection: - connection.execute( - text( - "INSERT INTO vectors.memory " - "(record_key, kind, content_hash, metadata, embedding) " - "VALUES ('forbidden', 'memory', 'x', '{}', '[1,0]')" - ) - ) - with pytest.raises(ProgrammingError): - with writer_engine.connect() as connection: - connection.execute( - text( - "SELECT metadata, 1 - (embedding <=> '[1,0]'::vector) AS similarity " - "FROM vectors.memory ORDER BY embedding <=> '[1,0]'::vector LIMIT 1" - ) - ) - reader_engine.dispose() - writer_engine.dispose() - - -def test_pgvector_writer_health_requires_sequence_usage(vector_configs): - from tht.adapters.vector.pgvector import PgVectorStore - - admin_config, _, _, no_sequence_config = vector_configs - store = PgVectorStore(None, no_sequence_config, expected_dimension=2) - - health = store.health() - assert health.ok is False - assert health.write_reachable is False - assert health.write_detail == ( - "vector schema incomplete: missing sequence privileges evidence, memory, schema_records" - ) - with pytest.raises(VectorWriteUnavailable) as error: - store.upsert("memory", [_record("needs-sequence", [1.0, 0.0])]) - assert isinstance(error.value.__cause__, InsufficientPrivilege) - - admin_engine = create_engine( - f"postgresql+psycopg2://{admin_config.user}:{admin_config.password}" - f"@{admin_config.host}:{admin_config.port}/{admin_config.database}" - ) - with admin_engine.begin() as connection: - connection.exec_driver_sql( - "GRANT USAGE ON ALL SEQUENCES IN SCHEMA vectors TO vector_l0_no_sequence" - ) - admin_engine.dispose() - - assert store.health().ok is True - assert store.upsert("memory", [_record("has-sequence", [1.0, 0.0])]) == 1 - - -def test_pgvector_health_requires_schema_usage_for_reader_and_writer(vector_configs): - from tht.adapters.vector.pgvector import PgVectorStore - - admin_config, reader_config, writer_config, _ = vector_configs - admin_engine = create_engine( - f"postgresql+psycopg2://{admin_config.user}:{admin_config.password}" - f"@{admin_config.host}:{admin_config.port}/{admin_config.database}" - ) - store = PgVectorStore(reader_config, writer_config, expected_dimension=2) - with admin_engine.begin() as connection: - connection.exec_driver_sql( - f"REVOKE USAGE ON SCHEMA vectors FROM {reader_config.user}, {writer_config.user}" - ) - health = store.health() - assert health.read_reachable is False and health.write_reachable is False - assert "missing schema usage" in health.read_detail - assert "missing schema usage" in health.write_detail - with pytest.raises(VectorReadUnavailable, match="Vector read operation unavailable"): - store.search(["memory"], [1.0, 0.0], limit=1) - with pytest.raises(VectorWriteUnavailable, match="Vector write operation unavailable"): - store.upsert("memory", [_record("blocked", [1.0, 0.0])]) - with admin_engine.begin() as connection: - connection.exec_driver_sql( - f"GRANT USAGE ON SCHEMA vectors TO {reader_config.user}, {writer_config.user}" - ) - admin_engine.dispose() - assert store.health().ok is True - - -def test_pgvector_maps_unavailable_connections_without_leaking_password(vector_configs): - from tht.adapters.vector.pgvector import PgVectorStore - - _, reader_config, writer_config, _ = vector_configs - password = "never-leak-this" - reader = reader_config.model_copy(update={"port": 1, "password": password}) - writer = writer_config.model_copy(update={"port": 1, "password": password}) - with pytest.raises(VectorReadUnavailable) as read_error: - PgVectorStore(reader, None).search(["memory"], [1.0, 0.0], limit=1) - with pytest.raises(VectorWriteUnavailable) as hash_error: - PgVectorStore(None, writer).existing_hashes("memory", ["memory"]) - with pytest.raises(VectorWriteUnavailable) as write_error: - PgVectorStore(None, writer).upsert("memory", [_record("x", [1.0, 0.0])]) - assert password not in str(read_error.value) - assert password not in str(hash_error.value) - assert password not in str(write_error.value) - - -def test_pgvector_health_reports_dimension_and_each_connection(vector_configs): - from tht.adapters.vector.pgvector import PgVectorStore - - _, reader_config, writer_config, _ = vector_configs - health = PgVectorStore(reader_config, writer_config, expected_dimension=2).health() - assert health.ok is True - assert health.read_reachable is True - assert health.write_reachable is True - assert health.observed_dimensions == (2,) - assert health.dimension_compatible is True - - mismatch = PgVectorStore(reader_config, None, expected_dimension=3).health() - assert mismatch.ok is False - assert mismatch.read_reachable is False - assert mismatch.read_detail == ( - "embedding dimension mismatch: evidence=2, memory=2, schema_records=2" - ) - assert mismatch.dimension_compatible is False - - -def test_pgvector_health_rejects_clean_and_partial_schemas(vector_configs): - from tht.adapters.vector.pgvector import PgVectorStore - - admin_config, _, _, _ = vector_configs - engine = create_engine( - f"postgresql+psycopg2://{admin_config.user}:{admin_config.password}" - f"@{admin_config.host}:{admin_config.port}/{admin_config.database}" - ) - with engine.begin() as connection: - connection.exec_driver_sql("CREATE SCHEMA clean_vectors") - connection.exec_driver_sql("CREATE SCHEMA partial_vectors") - connection.exec_driver_sql( - "CREATE TABLE partial_vectors.memory " - "(record_key text, kind text, content_hash text, metadata jsonb)" - ) - engine.dispose() - - clean = PgVectorStore( - admin_config.model_copy(update={"db_schema": "clean_vectors"}), - expected_dimension=2, - ).health() - assert clean.ok is False - assert clean.read_reachable is False - assert clean.read_detail == ( - "vector schema incomplete: missing tables evidence, memory, schema_records" - ) - - partial = PgVectorStore( - admin_config.model_copy(update={"db_schema": "partial_vectors"}), - expected_dimension=2, - ).health() - assert partial.ok is False - assert partial.read_reachable is False - assert partial.read_detail == ( - "vector schema incomplete: missing tables evidence, schema_records; " - "missing embedding columns memory" - ) diff --git a/harness/tests/l0/test_vector_adapter_parity.py b/harness/tests/l0/test_vector_adapter_parity.py deleted file mode 100644 index c77f21b1..00000000 --- a/harness/tests/l0/test_vector_adapter_parity.py +++ /dev/null @@ -1,294 +0,0 @@ -import math - -import pytest -from sqlalchemy import create_engine -from testcontainers.postgres import PostgresContainer - -from tht.adapters.vector.pgvector import PgVectorStore -from tht.adapters.vector.thoth_http import ThothHttpVectorStore -from tht.config import DatabaseConfig, RestConfig -from tht.ports.vector import VectorRecord, VectorStoreError, VectorWriteRecord -from tht.vectorstore.rest_client import VectorRestClient, VectorRestError - - -def _write(record_id, kind, embedding, content_hash): - return VectorWriteRecord( - VectorRecord( - id=record_id, - kind=kind, - ref="fixture", - title=record_id, - content=f"content {record_id}", - metadata={"fixture": True}, - ), - embedding, - content_hash, - ) - - -FIXTURE = [ - _write("memory:a", "memory", [1.0, 0.0], "hash-a"), - _write("memory:b", "memory", [1.0, 0.0], "hash-b"), - _write("solved:a", "solved_question", [0.8, 0.2], "hash-solved"), -] - - -class Response: - def __init__(self, payload=None, status=200): - self.status_code = status - self.payload = payload - self.text = "" if payload is None else "json" - - @property - def ok(self): - return self.status_code < 400 - - def json(self): - return self.payload - - -class FixtureHttpTransport: - def __init__(self): - self.rows = {} - self.calls = [] - - def post(self, url, json, headers, **kwargs): - assert headers == {"X-API-Key": "parity-key"} - self.calls.append((url.rsplit("/", 1)[-1], json)) - function = self.calls[-1][0] - if function == "list_tables": - return Response([{"table_name": "memory", "vector_dimensions": 2}]) - if function == "upsert_vector_records": - for row in json["rows"]: - self.rows[(json["table_name"], row["record_key"])] = row - return Response({"upserted": len(json["rows"])}) - if function == "existing_vector_hashes": - return Response([ - {"record_key": row["record_key"], "content_hash": row["content_hash"]} - for (table, _), row in self.rows.items() - if table == json["table_name"] and row["kind"] in json["kinds"] - ]) - assert function == "search_similar" - table_name = json["table_name"] - embedding = json["query_embedding"] - kinds = json.get("kinds") - - def similarity(row): - left, right = row["embedding"], embedding - return sum(a * b for a, b in zip(left, right)) / ( - math.sqrt(sum(a * a for a in left)) - * math.sqrt(sum(b * b for b in right)) - ) - - rows = [ - {"metadata": row["metadata"], "similarity": similarity(row)} - for (table, _), row in self.rows.items() - if table == table_name and (not kinds or row["kind"] in kinds) - ] - payload = sorted( - rows, - key=lambda row: (-row["similarity"], row["metadata"]["record_key"]), - )[: json["limit_count"]] - return Response(payload) - - -@pytest.fixture -def direct_store(): - with PostgresContainer("pgvector/pgvector:pg16") as postgres: - config = DatabaseConfig( - host=postgres.get_container_host_ip(), - port=int(postgres.get_exposed_port(5432)), - database=postgres.dbname, - schema="vectors", - user=postgres.username, - password=postgres.password, - ) - engine = create_engine(postgres.get_connection_url()) - with engine.begin() as connection: - connection.exec_driver_sql("CREATE SCHEMA vectors") - connection.exec_driver_sql("CREATE EXTENSION vector WITH SCHEMA vectors") - connection.exec_driver_sql( - "CREATE TABLE vectors.memory (" - "id bigserial PRIMARY KEY, record_key text UNIQUE NOT NULL, " - "kind text NOT NULL, content_hash text NOT NULL, metadata jsonb NOT NULL, " - "embedding vectors.vector(2) NOT NULL, indexed_at timestamptz NOT NULL " - "DEFAULT now())" - ) - engine.dispose() - reader, writer = config, config - store = PgVectorStore(reader, writer, expected_dimension=2) - store.upsert("memory", FIXTURE) - yield store - - -@pytest.fixture -def http_store(monkeypatch): - transport = FixtureHttpTransport() - monkeypatch.setattr("tht.vectorstore.rest_client.requests.post", transport.post) - client = VectorRestClient(RestConfig(base_url="https://vectors.test", api_key="parity-key")) - store = ThothHttpVectorStore(client, client, expected_dimension=2) - store.upsert("memory", FIXTURE) - store.transport = transport - return store - - -@pytest.mark.parametrize("store_fixture", ["direct_store", "http_store"]) -def test_kind_filtered_search_has_identical_order(request, store_fixture): - store = request.getfixturevalue(store_fixture) - hits = store.search(["memory"], [1.0, 0.0], limit=3, kinds=["memory"]) - assert [(hit.id, hit.kind, round(hit.similarity, 6)) for hit in hits] == [ - ("memory:a", "memory", 1.0), - ("memory:b", "memory", 1.0), - ] - - -@pytest.mark.parametrize("store_fixture", ["direct_store", "http_store"]) -def test_hash_and_upsert_parity(request, store_fixture): - store = request.getfixturevalue(store_fixture) - assert store.existing_hashes("memory", ["memory"]) == { - "memory:a": "hash-a", - "memory:b": "hash-b", - } - replacement = _write("memory:a", "memory", [0.0, 1.0], "hash-a-2") - assert store.upsert("memory", [replacement]) == 1 - assert store.existing_hashes("memory", ["memory"])["memory:a"] == "hash-a-2" - assert store.search(["memory"], [0.0, 1.0], limit=1, kinds=["memory"])[0].id == "memory:a" - - -@pytest.mark.parametrize("store_fixture", ["direct_store", "http_store"]) -def test_validation_error_parity(request, store_fixture): - store = request.getfixturevalue(store_fixture) - with pytest.raises(VectorStoreError, match="Collection not allowed"): - store.search(["not_allowed"], [1.0, 0.0], limit=1) - with pytest.raises(VectorStoreError, match="Kind not allowed"): - store.search(["memory"], [1.0, 0.0], limit=1, kinds=["not_allowed"]) - - -@pytest.mark.parametrize("store_fixture", ["direct_store", "http_store"]) -def test_dimension_error_parity(request, store_fixture): - store = request.getfixturevalue(store_fixture) - with pytest.raises(VectorStoreError, match="Query embedding dimension"): - store.search(["memory"], [1.0], limit=1) - with pytest.raises(VectorStoreError, match="Embedding dimension"): - store.upsert("memory", [_write("bad", "memory", [1.0], "bad")]) - - -def test_http_parity_exercises_rpc_kinds_payload(http_store): - http_store.search(["memory"], [1.0, 0.0], limit=2, kinds=["memory"]) - search_calls = [payload for function, payload in http_store.transport.calls if function == "search_similar"] - assert search_calls[-1] == { - "query_embedding": [1.0, 0.0], - "limit_count": 2, - "table_name": "memory", - "kinds": ["memory"], - } - - -def test_http_adapter_maps_transport_error(monkeypatch): - monkeypatch.setattr( - "tht.vectorstore.rest_client.requests.post", - lambda *args, **kwargs: Response({"message": "server broke"}, status=500), - ) - client = VectorRestClient(RestConfig(base_url="https://vectors.test", api_key="parity-key")) - store = ThothHttpVectorStore(client, client, expected_dimension=2) - with pytest.raises(VectorStoreError, match="HTTP 500"): - store.search(["memory"], [1.0, 0.0], limit=1, kinds=["memory"]) - - -def test_http_adapter_tolerates_malformed_metadata(monkeypatch): - monkeypatch.setattr( - "tht.vectorstore.rest_client.requests.post", - lambda *args, **kwargs: Response([{"similarity": 0.5, "metadata": None}]), - ) - client = VectorRestClient(RestConfig(base_url="https://vectors.test", api_key="parity-key")) - hit = ThothHttpVectorStore(client, None, expected_dimension=2).search( - ["memory"], [1.0, 0.0], limit=1 - )[0] - assert (hit.id, hit.kind, hit.metadata) == ("", "", {}) - - -def test_http_adapter_legacy_fallback_preserves_kind_semantics(monkeypatch): - calls = [] - - def post(url, json, **kwargs): - calls.append(json) - if "kinds" in json: - return Response({"message": "function not found"}, status=404) - return Response([ - {"similarity": 1.0, "metadata": {"record_key": "wrong", "kind": "solved_question"}}, - {"similarity": 0.9, "metadata": {"record_key": "right", "kind": "memory"}}, - ]) - - monkeypatch.setattr("tht.vectorstore.rest_client.requests.post", post) - client = VectorRestClient(RestConfig(base_url="https://vectors.test", api_key="parity-key")) - hits = ThothHttpVectorStore(client, None, expected_dimension=2).search( - ["memory"], [1.0, 0.0], limit=2, kinds=["memory"] - ) - assert [hit.id for hit in hits] == ["right"] - assert "kinds" in calls[0] and "kinds" not in calls[1] - - -def test_http_delete_generation_uses_exact_allowlisted_rpc_payload(monkeypatch): - calls = [] - monkeypatch.setattr( - "tht.vectorstore.rest_client.requests.post", - lambda url, json, **kwargs: calls.append((url, json)) or Response({"deleted": 2}), - ) - client = VectorRestClient(RestConfig(base_url="https://vectors.test", api_key="writer")) - assert client.delete_generation("evidence", "gen:" + "a" * 32, "default") == 2 - assert calls == [("https://vectors.test/rpc/delete_vector_generation", { - "table_name": "evidence", "kind": "evidence", "generation": "gen:" + "a" * 32, - "workspace_id": "default", - })] - - -def test_http_delete_generation_legacy_404_fails_closed_without_body_leak(monkeypatch): - monkeypatch.setattr( - "tht.vectorstore.rest_client.requests.post", - lambda *args, **kwargs: Response({"message": "secret legacy endpoint detail"}, status=404), - ) - client = VectorRestClient(RestConfig(base_url="https://vectors.test", api_key="writer")) - with pytest.raises(VectorRestError, match="delete_vector_generation RPC is unavailable") as error: - client.delete_generation("evidence", "gen:" + "a" * 32, "default") - assert "secret" not in str(error.value) - - -def test_http_rest_client_does_not_advertise_nonexistent_delete_kinds_rpc(): - client = VectorRestClient(RestConfig(base_url="https://vectors.test", api_key="writer")) - - assert hasattr(client, "delete_kinds") is False - - -def test_http_list_evidence_generations_exact_rpc_and_legacy_fail_closed(monkeypatch): - calls = [] - monkeypatch.setattr( - "tht.vectorstore.rest_client.requests.post", - lambda url, json, **kwargs: calls.append((url, json)) or Response([ - {"generation": "gen:" + "a" * 32} - ]), - ) - client = VectorRestClient(RestConfig(base_url="https://vectors.test", api_key="writer")) - assert client.list_evidence_generations("evidence", "default") == ["gen:" + "a" * 32] - assert calls[0][0].endswith("/rpc/list_evidence_generations") - assert calls[0][1] == {"table_name": "evidence", "kind": "evidence", "workspace_id": "default"} - - -@pytest.mark.parametrize("generation", ["gen:a", "gen:" + "A" * 32, "gen:" + "a" * 33]) -def test_http_generation_operations_reject_noncanonical_values(monkeypatch, generation): - monkeypatch.setattr( - "tht.vectorstore.rest_client.requests.post", - lambda *args, **kwargs: pytest.fail("invalid generation reached transport"), - ) - client = VectorRestClient(RestConfig(base_url="https://vectors.test", api_key="writer")) - with pytest.raises(ValueError, match="canonical"): - client.delete_generation("evidence", generation, "default") - - -def test_http_inventory_rejects_malformed_rpc_output(monkeypatch): - monkeypatch.setattr( - "tht.vectorstore.rest_client.requests.post", - lambda *args, **kwargs: Response([{"generation": "gen:../escape"}]), - ) - client = VectorRestClient(RestConfig(base_url="https://vectors.test", api_key="writer")) - with pytest.raises(VectorRestError, match="malformed"): - client.list_evidence_generations("evidence", "default") diff --git a/harness/tests/l0/test_vector_migrations.py b/harness/tests/l0/test_vector_migrations.py deleted file mode 100644 index 72640003..00000000 --- a/harness/tests/l0/test_vector_migrations.py +++ /dev/null @@ -1,303 +0,0 @@ -import json -from pathlib import Path - -import pytest -from sqlalchemy import create_engine, text -from sqlalchemy.exc import ProgrammingError -from testcontainers.postgres import PostgresContainer -from typer.testing import CliRunner - -from tht.cli import app -from tht.config import DatabaseConfig -from tht.ports.vector import VectorRecord, VectorWriteRecord - - -@pytest.fixture(scope="module") -def database_url(): - with PostgresContainer("pgvector/pgvector:pg16") as postgres: - yield postgres.get_connection_url() - - -def test_migrations_are_clean_and_idempotent(database_url): - from tht.cli.vector_migrate_cmd import migrate, migration_status - - before = migration_status(database_url) - assert [item.version for item in before.pending] == ["001", "002", "003", "004"] - - migrate(database_url) - migrate(database_url) - - status = migration_status(database_url) - assert status.pending == () - assert status.drifted == () - assert [item.version for item in status.applied] == ["001", "002", "003", "004"] - - -def test_schema_matches_direct_adapter_contract(database_url): - engine = create_engine(database_url) - with engine.connect() as connection: - rows = connection.execute( - text( - "SELECT table_name, column_name, data_type, udt_name " - "FROM information_schema.columns WHERE table_schema = 'vectors' " - "ORDER BY table_name, ordinal_position" - ) - ).all() - vector_types = connection.execute( - text( - "SELECT c.relname, format_type(a.atttypid, a.atttypmod) " - "FROM pg_class c JOIN pg_namespace n ON n.oid = c.relnamespace " - "JOIN pg_attribute a ON a.attrelid = c.oid AND a.attname = 'embedding' " - "WHERE n.nspname = 'vectors' ORDER BY c.relname" - ) - ).all() - engine.dispose() - - tables = {row.table_name for row in rows} - assert tables == {"evidence", "memory", "schema_records"} - required = {"id", "record_key", "kind", "content_hash", "metadata", "embedding", "indexed_at"} - for table in tables: - assert {row.column_name for row in rows if row.table_name == table} == required - assert vector_types == [ - ("evidence", "vectors.vector(768)"), - ("memory", "vectors.vector(768)"), - ("schema_records", "vectors.vector(768)"), - ] - - -def test_roles_have_runtime_privileges_only(database_url): - from tht.cli.vector_migrate_cmd import migrate - - migrate(database_url) - admin = create_engine(database_url) - with admin.begin() as connection: - connection.exec_driver_sql("ALTER ROLE vector_reader LOGIN PASSWORD 'reader-test-only'") - connection.exec_driver_sql("ALTER ROLE vector_writer LOGIN PASSWORD 'writer-test-only'") - url = admin.url - reader = create_engine(url.set(username="vector_reader", password="reader-test-only")) - writer = create_engine(url.set(username="vector_writer", password="writer-test-only")) - - from tht.adapters.vector.pgvector import PgVectorStore - - common = { - "host": url.host, - "port": url.port, - "database": url.database, - "schema": "vectors", - } - reader_config = DatabaseConfig( - **common, user="vector_reader", password="reader-test-only" - ) - writer_config = DatabaseConfig( - **common, user="vector_writer", password="writer-test-only" - ) - store = PgVectorStore(reader_config, writer_config, expected_dimension=768) - assert store.health().ok is True - assert store.upsert( - "memory", - [ - VectorWriteRecord( - record=VectorRecord( - id="adapter-write", - kind="memory", - ref="session:test", - title="test", - content="test", - ), - embedding=[0.0] * 768, - content_hash="adapter-hash", - ) - ], - ) == 1 - - with reader.connect() as connection: - connection.execute(text("SELECT metadata, embedding FROM vectors.memory")).all() - with pytest.raises(ProgrammingError): - with reader.begin() as connection: - connection.execute( - text( - "INSERT INTO vectors.memory " - "(record_key, kind, content_hash, metadata, embedding) " - "VALUES ('reader-write', 'memory', 'x', '{}', " - "array_fill(0, ARRAY[768])::vectors.vector)" - ) - ) - - with writer.begin() as connection: - connection.execute( - text( - "INSERT INTO vectors.memory " - "(record_key, kind, content_hash, metadata, embedding) " - "VALUES ('writer-ok', 'memory', 'x', '{}', " - "array_fill(0, ARRAY[768])::vectors.vector)" - ) - ) - assert connection.execute( - text("SELECT content_hash FROM vectors.memory WHERE record_key = 'writer-ok'") - ).scalar_one() == "x" - connection.execute( - text("UPDATE vectors.memory SET content_hash = 'y' WHERE record_key = 'writer-ok'") - ) - with pytest.raises(ProgrammingError): - with writer.connect() as connection: - connection.execute(text("SELECT metadata FROM vectors.memory")).all() - with pytest.raises(ProgrammingError): - with writer.begin() as connection: - connection.execute(text("DELETE FROM vectors.memory WHERE record_key = 'writer-ok'")) - - reader.dispose() - writer.dispose() - admin.dispose() - - -def test_status_json_is_pristine(database_url, monkeypatch): - monkeypatch.setenv("THT_VECTOR_ADMIN_URL", database_url) - result = CliRunner().invoke(app, ["vector", "migrate", "--status", "--json"]) - - assert result.exit_code == 0, result.output - assert json.loads(result.stdout) == { - "applied": ["001", "002", "003", "004"], - "drifted": [], - "pending": [], - } - assert result.stderr == "" - - -def test_checksum_drift_is_reported_and_refused(database_url, tmp_path): - from tht.cli.vector_migrate_cmd import MigrationError, migrate, migration_status - - migrations = _copy_migrations(tmp_path) - migrate(database_url, migrations) - (migrations / "002_schema_tables.sql").write_text("SELECT 2;\n") - - assert [item.version for item in migration_status(database_url, migrations).drifted] == [ - "002" - ] - with pytest.raises(MigrationError, match="checksum drift"): - migrate(database_url, migrations) - - -def test_unknown_applied_version_is_downgrade_drift(database_url): - from tht.cli.vector_migrate_cmd import MigrationError, migrate, migration_status - - migrate(database_url) - engine = create_engine(database_url) - with engine.begin() as connection: - connection.execute( - text( - "INSERT INTO public.tht_vector_migrations (version, name, checksum) " - "VALUES ('999', 'future', 'future-checksum'), " - "('future_x', 'future_named', 'future-checksum')" - ) - ) - try: - with pytest.raises( - MigrationError, match="absent from local manifest: 999, future_x" - ): - migration_status(database_url) - with pytest.raises( - MigrationError, match="absent from local manifest: 999, future_x" - ): - migrate(database_url) - finally: - with engine.begin() as connection: - connection.execute( - text( - "DELETE FROM public.tht_vector_migrations " - "WHERE version IN ('999', 'future_x')" - ) - ) - engine.dispose() - - -def test_migration_versions_sort_numerically_and_reject_numeric_duplicates(tmp_path): - from tht.cli.vector_migrate_cmd import MigrationError, _discover - - migrations = tmp_path / "ordered" - migrations.mkdir() - (migrations / "10_tenth.sql").write_text("SELECT 10;\n") - (migrations / "2_second.sql").write_text("SELECT 2;\n") - assert [item.version for item in _discover(migrations)] == ["2", "10"] - - (migrations / "02_duplicate.sql").write_text("SELECT 2;\n") - with pytest.raises(MigrationError, match="Duplicate migration version: 2"): - _discover(migrations) - - -def test_hostile_admin_search_path_cannot_shadow_migration_objects(database_url): - from tht.cli.vector_migrate_cmd import migrate - - admin = create_engine(database_url, isolation_level="AUTOCOMMIT") - with admin.connect() as connection: - connection.exec_driver_sql("DROP DATABASE IF EXISTS vector_hostile") - connection.exec_driver_sql("CREATE DATABASE vector_hostile") - hostile_url = admin.url.set(database="vector_hostile") - hostile = create_engine(hostile_url) - try: - with hostile.begin() as connection: - connection.exec_driver_sql("CREATE SCHEMA shadow") - connection.exec_driver_sql( - "CREATE TABLE shadow.tht_vector_migrations " - "(version text, checksum text, poisoned boolean DEFAULT true)" - ) - connection.exec_driver_sql("ALTER ROLE test SET search_path = shadow, public") - hostile.dispose() - - migrate(hostile_url.render_as_string(hide_password=False)) - - verification = create_engine(hostile_url) - with verification.connect() as connection: - assert connection.execute( - text("SELECT count(*) FROM public.tht_vector_migrations") - ).scalar_one() == 4 - assert connection.execute( - text("SELECT count(*) FROM shadow.tht_vector_migrations") - ).scalar_one() == 0 - assert connection.execute( - text( - "SELECT format_type(a.atttypid, a.atttypmod) " - "FROM pg_catalog.pg_attribute a " - "WHERE a.attrelid = 'vectors.memory'::pg_catalog.regclass " - "AND a.attname = 'embedding'" - ) - ).scalar_one() == "vectors.vector(768)" - verification.dispose() - finally: - cleanup = create_engine(database_url, isolation_level="AUTOCOMMIT") - with cleanup.connect() as connection: - connection.exec_driver_sql("ALTER ROLE test RESET search_path") - connection.exec_driver_sql( - "SELECT pg_catalog.pg_terminate_backend(pid) FROM pg_catalog.pg_stat_activity " - "WHERE datname = 'vector_hostile' AND pid <> pg_catalog.pg_backend_pid()" - ) - connection.exec_driver_sql("DROP DATABASE IF EXISTS vector_hostile") - cleanup.dispose() - admin.dispose() - - -def test_failed_batch_rolls_back_schema_and_ledger(database_url, tmp_path): - from tht.cli.vector_migrate_cmd import MigrationError, migrate, migration_status - - migrations = _copy_migrations(tmp_path) - (migrations / "005_first.sql").write_text("CREATE TABLE public.must_rollback (id int);\n") - (migrations / "006_broken.sql").write_text("THIS IS NOT SQL;\n") - - with pytest.raises(MigrationError, match="006_broken.sql"): - migrate(database_url, migrations) - - engine = create_engine(database_url) - with engine.connect() as connection: - assert connection.execute(text("SELECT to_regclass('public.must_rollback')")).scalar() is None - engine.dispose() - status = migration_status(database_url, migrations) - assert [item.version for item in status.applied] == ["001", "002", "003", "004"] - assert [item.version for item in status.pending] == ["005", "006"] - - -def _copy_migrations(tmp_path: Path) -> Path: - source = Path(__file__).parents[2] / "tht" / "migrations" / "vector" - target = tmp_path / "migrations" - target.mkdir() - for migration in source.glob("*.sql"): - (target / migration.name).write_bytes(migration.read_bytes()) - return target diff --git a/harness/tests/l2/test_memory_save_one_real.py b/harness/tests/l2/test_memory_save_one_real.py deleted file mode 100644 index 2399c496..00000000 --- a/harness/tests/l2/test_memory_save_one_real.py +++ /dev/null @@ -1,53 +0,0 @@ -"""L2: tht memory save-one against real pgvector (spec D11, L2). - -Validates D11 end-to-end: a single promoted decision is upserted to the real -pgvector via the WRITER key (not a full resync), and a subsequent search_similar -finds the memory. L1 tested the pure save_one_memory core; here the REST writer + -real pgvector + real embeddings are in the loop. - -Run: pytest -m l2 tests/l2/test_memory_save_one_real.py -s (needs .env + VPN + Ollama) -""" -from datetime import datetime -from pathlib import Path - -import pytest - -from tht.memory import MemoryRecord, save_one_memory -from tht.workspace import load_workspace - -pytestmark = [pytest.mark.l2] -WORKSPACE = Path(__file__).resolve().parents[2] / "workspaces" / "tht-test.yaml" - - -def test_save_one_upserts_to_real_pgvector(l2_env): - """save_one_memory pushes one row to the real pgvector via the writer key, and - a subsequent search_similar retrieves it. Idempotent (re-running upserts >= 0).""" - from tht.vectorstore.embeddings import OllamaEmbeddings - from tht.vectorstore.rest_client import VectorRestClient - - ws = load_workspace(WORKSPACE) - if not ws.vector_write_rest or not ws.vector_write_rest.api_key.strip(): - pytest.skip("vector_write_rest not configured (no writer key)") - - writer = VectorRestClient(ws.vector_write_rest) - embedder = OllamaEmbeddings(ws.embeddings) - - record = MemoryRecord( - id="mem-l2test", ts=datetime.now(), session_id="l2-self-test", - decision_seq=999, type="concept_clarified", subject="ablazione recente", - detail="evento di ablazione negli ultimi 15 anni", - rationale="L2 self-test (idempotent)", - question_context="ablazione 2025", tables=[], concepts=["ablazione recente"], - ) - from tht.adapters.vector import ThothHttpVectorStore - - store = ThothHttpVectorStore(reader=writer, writer=writer) - upserted = save_one_memory([record], decision_seq=999, store=store, embedder=embedder) - assert upserted >= 0 # idempotent: 0 on unchanged, >=1 on new/updated - - # read it back via the READER key (vector_rest, path /vector/v1/) - reader = VectorRestClient(ws.vector_rest) - qvec = embedder.embed_query("ablazione") - hits = reader.search_similar("memory", qvec, 10) - ids = {h.get("metadata", {}).get("record_key", "") for h in hits} - assert "memory:mem-l2test" in ids, "upserted memory not retrievable via search_similar" diff --git a/harness/tests/test_adapter_factory.py b/harness/tests/test_adapter_factory.py index 13fb5c54..d39b037a 100644 --- a/harness/tests/test_adapter_factory.py +++ b/harness/tests/test_adapter_factory.py @@ -2,11 +2,11 @@ import pytest from tht.adapters.dwh import PostgresDwhAdapter, ThothRestDwhAdapter from tht.adapters.factory import build_dwh, build_vector_store -from tht.adapters.vector import PgVectorStore, QdrantVectorStore, ThothHttpVectorStore +from tht.adapters.vector import QdrantVectorStore from tht.config import Config, ConfigError -def _config(*, dwh_type="thoth_rest", vector_type="thoth_vector_http", reader=True, writer=True): +def _config(*, dwh_type="thoth_rest", include_vectors=True): dwh = ( { "type": "thoth_rest", @@ -27,48 +27,12 @@ def _config(*, dwh_type="thoth_rest", vector_type="thoth_vector_http", reader=Tr ) vectors = ( { - "type": "thoth_vector_http", - **( - {"reader": {"base_url": "https://vectors.test/", "api_key": "reader"}} - if reader - else {} - ), - **( - {"writer": {"base_url": "https://vectors.test/", "api_key": "writer"}} - if writer - else {} - ), - } - if vector_type == "thoth_vector_http" - else { - "type": "pgvector_direct", - **( - { - "reader": { - "host": "vector-db", - "database": "postgres", - "schema": "vectors", - "user": "reader", - "password": "secret", - } - } - if reader - else {} - ), - **( - { - "writer": { - "host": "vector-db", - "database": "postgres", - "schema": "vectors", - "user": "writer", - "password": "secret", - } - } - if writer - else {} - ), + "type": "qdrant", + "base_url": "http://qdrant:6333", + "collection": "psd-clinical", } + if include_vectors + else None ) legacy_database = ( dwh["connection"] @@ -80,7 +44,19 @@ def _config(*, dwh_type="thoth_rest", vector_type="thoth_vector_http", reader=Tr "transport": "rest", } ) - return Config.model_validate({"dwh": dwh, "vectors": vectors, "database": legacy_database}) + payload = {"dwh": dwh, "database": legacy_database} + if vectors is not None: + payload["vectors"] = vectors + payload["embeddings"] = { + "provider": "ollama_internal", + "base_url": "http://embedding:11434", + "model": "qwen3-embedding:0.6b", + "dim": 1024, + } + config = Config.model_validate(payload) + config._workspace_id = "psd-clinical" + config._workspace_revision = "a" * 40 + return config @pytest.mark.parametrize( @@ -91,76 +67,8 @@ def test_factory_selects_dwh_adapter(dwh_type, adapter_type): assert isinstance(build_dwh(_config(dwh_type=dwh_type)), adapter_type) -def test_factory_selects_http_vector_and_requires_writer(): - config = _config(writer=False) - - assert isinstance(build_vector_store(config), ThothHttpVectorStore) - with pytest.raises(ConfigError, match="writer"): - build_vector_store(config, require_write=True) - - -def test_factory_builds_writer_only_http_vector_when_write_is_required(): - config = _config(reader=False, writer=True) - - store = build_vector_store(config, require_write=True) - assert isinstance(store, ThothHttpVectorStore) - assert store.capabilities.search is False - assert store.capabilities.upsert is True - - -def test_factory_selects_direct_vector_store_and_requires_writer(): - config = _config(vector_type="pgvector_direct", writer=False) - - assert isinstance(build_vector_store(config), PgVectorStore) - with pytest.raises(ConfigError, match="writer"): - build_vector_store(config, require_write=True) - - -def test_factory_builds_writer_only_direct_vector_when_write_is_required(): - store = build_vector_store( - _config(vector_type="pgvector_direct", reader=False), require_write=True - ) - assert isinstance(store, PgVectorStore) - assert store.capabilities.search is False - assert store.capabilities.upsert is True - - def test_factory_selects_qdrant_for_schema_v3_runtime(): - config = Config.model_validate( - { - "dwh": { - "type": "postgres_direct", - "connection": { - "host": "db", - "database": "analytics", - "schema": "mart", - "user": "reader", - "password": "secret", - }, - }, - "database": { - "host": "db", - "database": "analytics", - "schema": "mart", - "user": "reader", - "password": "secret", - "transport": "direct", - }, - "vectors": { - "type": "qdrant", - "base_url": "http://qdrant:6333", - "collection": "psd-clinical", - }, - "embeddings": { - "provider": "ollama_internal", - "base_url": "http://embedding:11434", - "model": "qwen3-embedding:0.6b", - "dim": 1024, - }, - } - ) - config._workspace_id = "psd-clinical" - config._workspace_revision = "a" * 40 + config = _config(dwh_type="postgres_direct") store = build_vector_store(config, require_write=True) @@ -169,18 +77,9 @@ def test_factory_selects_qdrant_for_schema_v3_runtime(): assert store.capabilities.upsert is True -def test_factory_reuses_legacy_direct_connection_for_server_writes_only(): - server = _config(vector_type="pgvector_direct", writer=False) - server.vectors.connection = server.vectors.reader - server.vectors.reader = None - - store = build_vector_store(server, require_write=True) - assert store.capabilities.search is True - assert store.capabilities.upsert is True - - server.profile = "workstation" - with pytest.raises(ConfigError, match="writer"): - build_vector_store(server, require_write=True) +def test_factory_requires_qdrant_vector_resource(): + with pytest.raises(ConfigError, match="vectors"): + build_vector_store(_config(include_vectors=False)) def test_factory_propagates_non_default_statement_timeout(): diff --git a/harness/tests/test_blocco6_robustness.py b/harness/tests/test_blocco6_robustness.py deleted file mode 100644 index d6db9f19..00000000 --- a/harness/tests/test_blocco6_robustness.py +++ /dev/null @@ -1,54 +0,0 @@ -"""Blocco 6: robustness fixes -- taskdoc slice/bound, report escaping, upsert count.""" -from tht.report import _markdown_table, extract_reviewer_notes -from tht.taskdoc import generate_task_doc - - -def test_taskdoc_slices_to_promoted_tables(tmp_path): - s = tmp_path / "sess" - s.mkdir() - (s / "question.md").write_text("q") - (s / "schema_linking.json").write_text( - '{"question":"q","candidates":[' - '{"kind":"table","name":"pazienti","decision":"promoted"},' - '{"kind":"table","name":"ricoveri","decision":"promoted"}],' - '"joins":[],"excluded":[],"open_questions":[]}' - ) - doc = generate_task_doc(session_dir=s, phase=4, promoted_tables=["pazienti"]) - assert "pazienti" in doc.body - assert "ricoveri" not in doc.body # sliced out - - -def test_taskdoc_truncates_over_budget(tmp_path): - s = tmp_path / "sess" - s.mkdir() - (s / "question.md").write_text("# Domanda\n" + "x" * 200_000) - doc = generate_task_doc(session_dir=s, phase=1) - assert doc.byte_budget_ok is False - assert len(doc.body.encode()) <= 80_000 - assert "troncato" in doc.body - - -def test_markdown_table_escapes_pipes_and_newlines(): - table = _markdown_table(["c"], [("a|b\nc",)]) - # the cell must not introduce a raw pipe or newline that breaks the row - body_line = table.splitlines()[2] - assert "\\|" in body_line - assert "\n" not in body_line - - -def test_extract_reviewer_notes_uses_last_heading(): - report = ( - "## Note del reviewer\nnella cella di dati appariva questo testo\n" - "## Note del reviewer\nnota vera del reviewer" - ) - assert extract_reviewer_notes(report) == "nota vera del reviewer" - - -def test_upsert_count_handles_postgrest_list_wrapping(): - from unittest.mock import MagicMock - - from tht.vectorstore.rest_client import VectorRestClient - - client = VectorRestClient.__new__(VectorRestClient) - client._call = MagicMock(return_value=[{"upserted": 7}]) # list-wrapped scalar - assert client.upsert_records("memory", [{}, {}]) == 7 diff --git a/harness/tests/test_config_legacy_compat.py b/harness/tests/test_config_legacy_compat.py deleted file mode 100644 index 2166ad09..00000000 --- a/harness/tests/test_config_legacy_compat.py +++ /dev/null @@ -1,148 +0,0 @@ -import json -import os -import subprocess -from pathlib import Path - -import pytest -from typer.testing import CliRunner - -from tht.cli import app -from tht.config import load_config -from tht.adapters.factory import build_vector_store - - -def _write_old_workspace(tmp_path): - path = tmp_path / "old.yaml" - path.write_text( - """ -database: - host: ignored-for-rest - database: analytics - schema: mart - user: legacy-user - password: legacy-password - transport: rest -rest: - base_url: https://dwh.example.test/ - api_key: dwh-reader -vector_db: - host: vector-db - database: postgres - schema: vectors - user: vector-user - password: vector-password -vector_rest: - base_url: https://vectors.example.test/ - api_key: vector-reader -vector_write_rest: - base_url: https://vectors.example.test/ - api_key: vector-writer -paths: - artifacts: build/artifacts - indexes: build/indexes - sessions: build/sessions -""" - ) - return path - - -def _write_new_workspace(tmp_path): - path = tmp_path / "new.yaml" - path.write_text( - """ -dwh: - type: thoth_rest - database: - database: analytics - schema: mart - endpoint: - base_url: https://dwh.example.test/ - api_key: dwh-reader -vectors: - type: thoth_vector_http - reader: - base_url: https://vectors.example.test/ - api_key: vector-reader - writer: - base_url: https://vectors.example.test/ - api_key: vector-writer - direct: - host: vector-db - database: postgres - schema: vectors - user: vector-user - password: vector-password -roots: - artifacts: build/artifacts - indexes: build/indexes - sessions: build/sessions -""" - ) - return path - - -def test_legacy_rest_workspace_equals_new_resource_schema(tmp_path, capsys): - with pytest.warns(FutureWarning, match="DEPRECATION") as warnings: - old = load_config(_write_old_workspace(tmp_path)) - captured = capsys.readouterr() - new = load_config(_write_new_workspace(tmp_path)) - - assert old.dwh.model_dump() == new.dwh.model_dump() - assert old.vectors.model_dump() == new.vectors.model_dump() - assert old.roots.model_dump() == new.roots.model_dump() - assert captured.out == "" - assert captured.err == "" - assert len(warnings) == 1 - - -def test_legacy_warning_does_not_contaminate_cli_json(tmp_path): - with pytest.warns(FutureWarning, match="DEPRECATION") as warnings: - result = CliRunner().invoke( - app, - ["session", "list", "--json", "-c", str(_write_old_workspace(tmp_path))], - ) - - assert result.exit_code == 0 - json.loads(result.stdout) - assert "DEPRECATION" not in result.stdout - assert result.stderr == "" - assert len(warnings) == 1 - - -def test_legacy_cli_subprocess_warns_once_on_stderr_and_keeps_json_stdout(tmp_path): - workspace = _write_old_workspace(tmp_path) - result = subprocess.run( - [ - str(Path(__file__).parents[1] / ".venv" / "bin" / "tht"), - "session", - "list", - "--json", - "-c", - str(workspace), - ], - cwd=tmp_path, - env={**os.environ, "PYTHONWARNINGS": "default"}, - text=True, - capture_output=True, - check=False, - ) - - assert result.returncode == 0 - json.loads(result.stdout) - assert "DEPRECATION" not in result.stdout - assert result.stderr.count("DEPRECATION") == 1 - - -def test_legacy_writer_only_vector_config_builds_for_targeted_writes(tmp_path): - workspace = _write_old_workspace(tmp_path) - content = workspace.read_text().replace( - "vector_rest:\n base_url: https://vectors.example.test/\n api_key: vector-reader\n", - "", - ) - workspace.write_text(content) - - with pytest.warns(FutureWarning): - cfg = load_config(workspace) - store = build_vector_store(cfg, require_write=True) - assert store.capabilities.search is False - assert store.capabilities.upsert is True diff --git a/harness/tests/test_search_similar_kinds.py b/harness/tests/test_search_similar_kinds.py deleted file mode 100644 index 049d9519..00000000 --- a/harness/tests/test_search_similar_kinds.py +++ /dev/null @@ -1,123 +0,0 @@ -"""L1: filtro `kinds` server-side su search_similar (fast-follow post active-memory). - -`memory` e `solved_question` condividono la tabella pgvector: senza filtro nel -`WHERE` della RPC, il top-k della tabella mista puo' affamare la ricerca memorie -(e viceversa) perche' il filtro per kind avveniva solo client-side DOPO il taglio -a top_n. Questi test fissano il contratto client: -- il client manda `kinds` nel payload della RPC quando richiesto (filtro esatto); -- su un server legacy (funzione a 3 argomenti -> PostgREST 404) ritenta senza - `kinds`, lasciando il filtro al post-filter client-side esistente; -- RestSearcher inoltra i kinds alla RPC. -""" -import pytest - -from tht.config import RestConfig -from tht.vectorstore.reader import RestSearcher -from tht.vectorstore.rest_client import VectorRestClient, VectorRestError - - -class _Resp: - def __init__(self, status_code=200, payload=None, text=""): - self.status_code = status_code - self._payload = [] if payload is None else payload - self.text = text or ("[]" if status_code == 200 else text) - - @property - def ok(self): - return self.status_code < 400 - - def json(self): - if not self.ok: - return {"message": self.text} - return self._payload - - -def _client() -> VectorRestClient: - return VectorRestClient(RestConfig(base_url="https://v/", api_key="K-READ")) - - -def test_search_similar_sends_kinds_in_rpc_payload(monkeypatch): - seen = [] - - def fake_post(url, json=None, **kw): - seen.append(json) - return _Resp(payload=[{"similarity": 0.9, "metadata": {"kind": "memory"}}]) - - monkeypatch.setattr("tht.vectorstore.rest_client.requests.post", fake_post) - rows = _client().search_similar("memory", [0.1] * 4, 5, kinds=["memory"]) - assert len(rows) == 1 - assert seen[0]["kinds"] == ["memory"] - assert seen[0]["table_name"] == "memory" - assert seen[0]["limit_count"] == 5 - - -def test_search_similar_omits_kinds_when_none(monkeypatch): - seen = [] - - def fake_post(url, json=None, **kw): - seen.append(json) - return _Resp() - - monkeypatch.setattr("tht.vectorstore.rest_client.requests.post", fake_post) - _client().search_similar("memory", [0.1] * 4, 5) - assert "kinds" not in seen[0] - - -def test_search_similar_falls_back_without_kinds_on_legacy_404(monkeypatch): - # Server legacy: la funzione a 4 argomenti non esiste -> PostgREST 404 (PGRST202). - # Il client ritenta senza `kinds`; il filtro resta al post-filter client-side. - seen = [] - - def fake_post(url, json=None, **kw): - seen.append(json) - if "kinds" in json: - return _Resp(status_code=404, text="Could not find the function (PGRST202)") - return _Resp(payload=[{"similarity": 0.8, "metadata": {"kind": "memory"}}]) - - monkeypatch.setattr("tht.vectorstore.rest_client.requests.post", fake_post) - rows = _client().search_similar("memory", [0.1] * 4, 5, kinds=["memory"]) - assert len(rows) == 1 - assert len(seen) == 2 - assert "kinds" in seen[0] and "kinds" not in seen[1] - - -def test_search_similar_reraises_non_404_with_kinds(monkeypatch): - def fake_post(url, json=None, **kw): - return _Resp(status_code=500, text="boom") - - monkeypatch.setattr("tht.vectorstore.rest_client.requests.post", fake_post) - with pytest.raises(VectorRestError, match="HTTP 500"): - _client().search_similar("memory", [0.1] * 4, 5, kinds=["memory"]) - - -def test_generation_filter_is_sent_exactly_and_legacy_404_fails_closed(monkeypatch): - calls = [] - - def fake_call(self, function, payload): - calls.append(payload) - raise VectorRestError("HTTP 404 missing filtered RPC") - - monkeypatch.setattr(VectorRestClient, "_call", fake_call) - metadata_filter = {"vector_generation": "gen:abc", "document_ids": ["doc:1"]} - with pytest.raises(VectorRestError, match="404"): - _client().search_similar( - "evidence", [0.1] * 4, 5, kinds=["evidence"], metadata_filter=metadata_filter - ) - assert calls == [{ - "query_embedding": [0.1] * 4, "limit_count": 5, "table_name": "evidence", - "kinds": ["evidence"], "metadata_filter": metadata_filter, - }] - - -def test_rest_searcher_forwards_kinds_to_client(): - calls = [] - - class FakeClient: - def search_similar(self, table_name, query_vec, top_n, kinds=None): - calls.append((table_name, top_n, kinds)) - return [{"similarity": 0.7, "metadata": {"kind": "solved_question", - "record_key": "solved:s1"}}] - - hits = RestSearcher(FakeClient()).search([0.1] * 4, top_n=3, kinds=["solved_question"]) - assert calls == [("memory", 3, ["solved_question"])] - assert [h.kind for h in hits] == ["solved_question"] diff --git a/harness/tests/test_solved_search_cli.py b/harness/tests/test_solved_search_cli.py index d1b27f91..0631ed25 100644 --- a/harness/tests/test_solved_search_cli.py +++ b/harness/tests/test_solved_search_cli.py @@ -13,8 +13,7 @@ from typer.testing import CliRunner from tht.cli import app from tht.memory import MemoryRecord, save_registry -from tht.ports.vector import VectorReadUnavailable -from tht.vectorstore.rest_client import VectorRestError +from tht.ports.vector import VectorReadUnavailable, VectorStoreError from tht.vectorstore.store import VectorHit @@ -31,7 +30,7 @@ def _cfg(tmp_path): def test_solved_search_degrades_when_vectordb_unreachable(tmp_path, monkeypatch): def boom(cfg): - raise VectorRestError("Vector REST non raggiungibile su https://v/ (rpc search_similar)") + raise VectorStoreError("Qdrant non raggiungibile") monkeypatch.setattr("tht.cli.vector_cmd.open_searcher", boom) res = CliRunner().invoke( @@ -57,7 +56,7 @@ def test_solved_search_degrades_direct_vector_read_error(tmp_path, monkeypatch): def test_solved_search_degrades_human_mode(tmp_path, monkeypatch): def boom(cfg): - raise VectorRestError("Vector REST non raggiungibile") + raise VectorStoreError("Qdrant non raggiungibile") monkeypatch.setattr("tht.cli.vector_cmd.open_searcher", boom) res = CliRunner().invoke( diff --git a/harness/tests/test_vector_dual_key.py b/harness/tests/test_vector_dual_key.py deleted file mode 100644 index 290a6551..00000000 --- a/harness/tests/test_vector_dual_key.py +++ /dev/null @@ -1,64 +0,0 @@ -"""L1: dual vector API key (spec D11, §5.4). - -The reader (search_similar) and the writer (upsert_vector_records) use SEPARATE -API keys against the same pgvector REST endpoint, with distinct roles -(vector_reader / vector_writer). This test pins the dual-key construction and -the workstation write-guard. -""" -from tht.cli._guards import has_vector_write_rest, require_vector_write_allowed -from tht.config import Config, DatabaseConfig, RestConfig -from tht.vectorstore.rest_client import VectorRestClient - - -def _minimal_config(**kw) -> Config: - base = dict( - database=DatabaseConfig(database="db", schema="dw", user="u", password="p"), - ) - base.update(kw) - return Config(**base) - - -def test_reader_and_writer_use_separate_keys(): - reader = VectorRestClient(RestConfig(base_url="https://v/", api_key="K-READ")) - writer = VectorRestClient(RestConfig(base_url="https://v/", api_key="K-WRITE")) - assert reader.api_key == "K-READ" - assert writer.api_key == "K-WRITE" - - -def test_has_vector_write_rest_false_for_empty_key(): - cfg = _minimal_config(vector_write_rest=RestConfig(base_url="x", api_key=" ")) - assert has_vector_write_rest(cfg) is False - - -def test_has_vector_write_rest_false_when_absent(): - cfg = _minimal_config() - assert has_vector_write_rest(cfg) is False - - -def test_has_vector_write_rest_true_when_key_present(): - cfg = _minimal_config(vector_write_rest=RestConfig(base_url="x", api_key="K-WRITE")) - assert has_vector_write_rest(cfg) is True - - -def test_require_vector_write_allowed_blocks_workstation_without_key(): - import typer - cfg = _minimal_config(profile="workstation") # no vector_write_rest - try: - require_vector_write_allowed(cfg, "memory save-one") - assert False, "should have exited with code 4" - except typer.Exit as e: - assert e.exit_code == 4 - - -def test_require_vector_write_allowed_allows_workstation_with_key(): - cfg = _minimal_config( - profile="workstation", - vector_write_rest=RestConfig(base_url="x", api_key="K-WRITE"), - ) - require_vector_write_allowed(cfg, "memory save-one") # no exit -> ok - - -def test_require_vector_write_allowed_allows_server_without_key(): - # server profile can use direct vectordb; the REST write guard does not apply. - cfg = _minimal_config(profile="server") - require_vector_write_allowed(cfg, "memory save-one") # no exit -> ok diff --git a/harness/tests/test_vector_migration_packaging.py b/harness/tests/test_vector_migration_packaging.py index 03addb39..e52608e4 100644 --- a/harness/tests/test_vector_migration_packaging.py +++ b/harness/tests/test_vector_migration_packaging.py @@ -6,7 +6,7 @@ import zipfile from pathlib import Path -def test_built_wheel_installs_migrations_and_discovers_cli(tmp_path): +def test_built_wheel_omits_vector_sql_migrations_and_discovers_cli(tmp_path): harness = Path(__file__).parents[1] wheelhouse = tmp_path / "wheelhouse" target = tmp_path / "site" @@ -31,8 +31,7 @@ def test_built_wheel_installs_migrations_and_discovers_cli(tmp_path): wheel = next(wheelhouse.glob("tht-*.whl")) with zipfile.ZipFile(wheel) as archive: names = set(archive.namelist()) - assert "tht/migrations/vector/001_extensions.sql" in names - assert "tht/migrations/vector/003_roles.sql" in names + assert not any(name.startswith("tht/migrations/vector/") for name in names) assert "tht/migrations/sessions/001_schema.sql" in names assert "tht/migrations/sessions/002_security.sql" in names @@ -47,11 +46,11 @@ def test_built_wheel_installs_migrations_and_discovers_cli(tmp_path): [ sys.executable, "-c", - "from typer.testing import CliRunner; from tht.cli import app; " - "r=CliRunner().invoke(app, ['vector','migrate','--help']); " - "assert r.exit_code == 0, r.output; " - "r=CliRunner().invoke(app, ['session','migrate','--help']); " - "print(r.output); raise SystemExit(r.exit_code)", + ( + "from typer.testing import CliRunner; from tht.cli import app; " + "r=CliRunner().invoke(app, ['session','migrate','--help']); " + "print(r.output); raise SystemExit(r.exit_code)" + ), ], env=env, check=False, diff --git a/harness/tests/test_vector_port_contract.py b/harness/tests/test_vector_port_contract.py index 5fae5236..ae7cd4c5 100644 --- a/harness/tests/test_vector_port_contract.py +++ b/harness/tests/test_vector_port_contract.py @@ -3,255 +3,78 @@ from unittest.mock import MagicMock import pytest -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, - VectorWriteRecord, - VectorWriteUnavailable, + VectorStoreError, ) -from tht.vectorstore.records import evidence_records - - -def test_http_store_reports_reader_without_writer(): - reader = MagicMock() - store = ThothHttpVectorStore(reader=reader, writer=None) - - assert store.capabilities.search is True - assert store.capabilities.upsert is False - with pytest.raises(VectorWriteUnavailable): - store.upsert("memory", []) - - -def test_http_store_supports_writer_without_reader(): - writer = MagicMock() - store = ThothHttpVectorStore(reader=None, writer=writer, expected_dimension=768) - - assert store.capabilities.search is False - assert store.capabilities.existing_hashes is True - assert store.capabilities.upsert is True - assert hasattr(store, "delete_kinds") is False - with pytest.raises(VectorReadUnavailable): - store.search(["memory"], [0.1], limit=1) - - -@pytest.mark.parametrize("limit", [True, False, 1.0, 0, -1]) -def test_http_search_requires_a_strict_positive_integer_limit(limit): - store = ThothHttpVectorStore(reader=MagicMock(), writer=None) - - with pytest.raises(ValueError, match="positive integer"): - store.search(["memory"], [0.1], limit=limit) - - -def test_http_store_keeps_reader_and_writer_operations_separate(): - reader = MagicMock() - reader.search_similar.return_value = [ - { - "similarity": 0.75, - "metadata": { - "record_key": "m1", - "kind": "memory", - "ref": "session:s1", - "title": "Choice", - "content": "Use the curated table", - }, - } - ] - writer = MagicMock() - writer.existing_hashes.return_value = {"m1": "abc"} - writer.upsert_records.return_value = 1 - store = ThothHttpVectorStore(reader=reader, writer=writer) - - hits = store.search(["memory"], [0.1, 0.2], limit=3, kinds=["memory"]) - assert hits == [ - VectorHit( - id="m1", - kind="memory", - ref="session:s1", - title="Choice", - content="Use the curated table", - metadata={ - "record_key": "m1", - "kind": "memory", - "ref": "session:s1", - "title": "Choice", - "content": "Use the curated table", - }, - similarity=0.75, - ) - ] - reader.search_similar.assert_called_once_with( - "memory", [0.1, 0.2], 3, kinds=["memory"] - ) - writer.search_similar.assert_not_called() - - assert store.existing_hashes("memory", ["memory"]) == {"m1": "abc"} - writer.existing_hashes.assert_called_once_with("memory", ["memory"]) - - records = [ - VectorWriteRecord( - record=VectorRecord( - id="m1", - kind="memory", - ref="session:s1", - title="Choice", - content="Use the curated table", - ), - embedding=[0.1, 0.2], - content_hash="abc", - ) - ] - assert store.upsert("memory", records) == 1 - writer.upsert_records.assert_called_once() - reader.upsert_records.assert_not_called() - - -def test_http_upsert_serializes_a_canonical_builder_record(): - record = evidence_records( - [EvidenceDoc(id="joins", title="Join guidance", body="Use the curated join")], - max_chunk_chars=1000, - )[0] - writer = MagicMock() - writer.upsert_records.return_value = 1 - store = ThothHttpVectorStore(reader=MagicMock(), writer=writer) - - assert store.upsert( - "evidence", - [VectorWriteRecord(record=record, embedding=[0.2, 0.3], content_hash="digest")], - ) == 1 - row = writer.upsert_records.call_args.args[1][0] - assert row["record_key"] == "evidence:joins:0" - assert row["metadata"]["status"] == "reviewed" - assert row["embedding"] == [0.2, 0.3] - assert row["content_hash"] == "digest" - - -def test_http_upsert_preserves_metadata_named_like_transport_fields(): - record = VectorRecord( - id="collision", - kind="memory", - ref="session:s1", - title="Collision", - content="Semantic metadata must survive", - metadata={"embedding": "semantic embedding", "content_hash": "semantic hash"}, - ) - writer = MagicMock() - store = ThothHttpVectorStore(reader=MagicMock(), writer=writer) - - store.upsert( - "memory", - [VectorWriteRecord(record=record, embedding=[0.4], content_hash="transport hash")], - ) - row = writer.upsert_records.call_args.args[1][0] - assert row["embedding"] == [0.4] - assert row["content_hash"] == "transport hash" - assert row["metadata"]["embedding"] == "semantic embedding" - assert row["metadata"]["content_hash"] == "semantic hash" - - -def test_http_store_is_runtime_vector_store(): - store = ThothHttpVectorStore(reader=MagicMock(), writer=None) - assert isinstance(store, VectorStore) 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 assert PublicQdrantStore is QdrantVectorStore - assert PublicHttpStore is ThothHttpVectorStore assert PublicVectorStore is VectorStore - assert PublicVectorWriteRecord is VectorWriteRecord assert PublicVectorReadUnavailable is VectorReadUnavailable - capabilities = store_capabilities = ThothHttpVectorStore( - reader=MagicMock(), writer=None + capabilities = store_capabilities = 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": {}, + } + }, + ), ).capabilities assert capabilities.search is True with pytest.raises(FrozenInstanceError): store_capabilities.search = False -def test_http_health_uses_reader_list_tables_and_reports_failure(): - reader = MagicMock() - store = ThothHttpVectorStore(reader=reader, writer=None) - assert store.health().ok is True - - reader.list_tables.side_effect = RuntimeError("offline") - health = store.health() - assert health.ok is False - assert health.detail == "offline" - - -def test_http_health_reports_read_write_and_dimension_status_independently(): - reader = MagicMock() - reader.list_tables.return_value = [ - {"table_name": "memory", "vector_dimensions": 768} - ] - writer = MagicMock() - writer.list_tables.return_value = [ - {"table_name": "memory", "vector_dimensions": 768} - ] - store = ThothHttpVectorStore(reader, writer, expected_dimension=768) - - health = store.health() - assert health.ok is True - assert health.read_configured is True - assert health.read_reachable is True - assert health.write_configured is True - assert health.write_reachable is True - assert health.expected_dimension == 768 - assert health.observed_dimensions == (768,) - assert health.dimension_compatible is True - - -def test_http_health_does_not_hide_writer_failure_behind_reader_success(): - reader = MagicMock() - reader.list_tables.return_value = [] - writer = MagicMock() - writer.list_tables.side_effect = RuntimeError("writer offline") - store = ThothHttpVectorStore(reader, writer, expected_dimension=768) - - health = store.health() - assert health.ok is False - assert health.read_reachable is True - assert health.write_reachable is False - assert health.write_detail == "writer offline" - assert health.dimension_compatible is None - - -def test_http_health_covers_read_only_and_write_only_configuration(): - reader = MagicMock() - reader.list_tables.return_value = [{"vector_dimensions": 384}] - read_health = ThothHttpVectorStore(reader, None, expected_dimension=768).health() - assert read_health.ok is False - assert read_health.write_configured is False - assert read_health.write_reachable is None - assert read_health.dimension_compatible is False - - writer = MagicMock() - writer.list_tables.return_value = [{"vector_dimensions": 768}] - write_health = ThothHttpVectorStore(None, writer, expected_dimension=768).health() - assert write_health.ok is True - assert write_health.read_configured is False - assert write_health.read_reachable is None - assert write_health.dimension_compatible is True - - @pytest.mark.parametrize("limit", [True, False, 1.0, 0, -1]) -def test_legacy_direct_search_requires_a_strict_positive_integer_limit(limit): - store = LegacyDirectVectorStore(engine=MagicMock()) +def test_qdrant_search_requires_a_strict_positive_integer_limit(limit): + 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":{"points":[]}}', + json=lambda: {"result": {"points": []}}, + ), + ) with pytest.raises(ValueError, match="positive integer"): - store.search(["memory"], [0.1], limit=limit) + store.search(["memory"], [0.1] * 1024, limit=limit) + + +def test_qdrant_search_rejects_dimension_mismatches_before_transport(): + seen = [] + store = QdrantVectorStore( + base_url="http://qdrant:6333", + collection="workspace-semantic", + workspace_id="demo", + expected_dimension=1024, + request=lambda *args, **kwargs: seen.append((args, kwargs)), + ) + + with pytest.raises(VectorStoreError, match="dimension"): + store.search(["memory"], [0.1], limit=1) + + assert seen == [] def test_qdrant_store_is_runtime_vector_store(): diff --git a/harness/tht/adapters/factory.py b/harness/tht/adapters/factory.py index 85e906f0..683489fc 100644 --- a/harness/tht/adapters/factory.py +++ b/harness/tht/adapters/factory.py @@ -3,12 +3,10 @@ from tht.adapters.dwh import PostgresDwhAdapter, ThothRestDwhAdapter from tht.adapters.evidence import FilesystemEvidenceSource, HttpManifestEvidenceSource from tht.adapters.evidence.s3 import S3EvidenceSource -from tht.adapters.vector import PgVectorStore, QdrantVectorStore, ThothHttpVectorStore +from tht.adapters.vector import QdrantVectorStore from tht.config import Config, ConfigError -from tht.db.connection import make_engine from tht.ports.dwh import DwhAdapter from tht.ports.vector import VectorStore -from tht.vectorstore.rest_client import VectorRestClient def build_dwh(cfg: Config) -> DwhAdapter: @@ -33,29 +31,6 @@ def build_vector_store(cfg: Config, *, require_write: bool = False) -> VectorSto raise ConfigError("Risorsa vectors non configurata") match resource.type: - case "pgvector_direct": - reader = resource.reader or resource.connection - # Legacy server workspaces use one RW `vector_db` connection. Keep - # that deployment contract without turning a workstation's legacy - # compatibility connection into an implicit writer. - writer = resource.writer or ( - resource.connection if cfg.profile == "server" else None - ) - if require_write and writer is None: - raise ConfigError("Vector writer non configurato per pgvector_direct") - return PgVectorStore( - reader, - writer, - expected_dimension=cfg.embeddings.dim if cfg.embeddings is not None else None, - ) - case "thoth_vector_http": - if require_write and resource.writer is None: - raise ConfigError("Vector writer non configurato") - return ThothHttpVectorStore( - VectorRestClient(resource.reader) if resource.reader is not None else None, - VectorRestClient(resource.writer) if resource.writer is not None else None, - expected_dimension=cfg.embeddings.dim if cfg.embeddings is not None else None, - ) case "qdrant": return QdrantVectorStore( base_url=resource.base_url, @@ -68,40 +43,6 @@ def build_vector_store(cfg: Config, *, require_write: bool = False) -> VectorSto raise ConfigError(f"Adapter vector non supportato: {other}") -def build_vector_loader(cfg: Config, collection: str): - """Compatibility construction for legacy collection sync commands.""" - resource = cfg.vectors - if resource is None: - raise ConfigError("Risorsa vectors non configurata") - if cfg.embeddings is None: - raise ConfigError("Embeddings non configurati") - - if ( - resource.type == "thoth_vector_http" - and resource.writer is not None - and (cfg.profile == "workstation" or resource.direct is None) - ): - from tht.vectorstore.rest_writer import RestVectorWriter - - return RestVectorWriter(VectorRestClient(resource.writer), table=collection) - - connection = ( - resource.writer or resource.connection - if resource.type == "pgvector_direct" - else resource.direct - ) - if connection is None: - raise ConfigError("Vector writer non configurato") - from tht.vectorstore.store import VectorStore as TableVectorStore - - return TableVectorStore( - make_engine(connection), - schema=connection.db_schema, - table=collection, - dim=cfg.embeddings.dim, - ) - - def build_evidence_sources(cfg: Config): """Build configured Evidence sources, including the legacy curated filesystem tree.""" evidence = cfg.evidence @@ -152,4 +93,4 @@ def build_evidence_sources(cfg: Config): return sources -__all__ = ["build_dwh", "build_evidence_sources", "build_vector_loader", "build_vector_store"] +__all__ = ["build_dwh", "build_evidence_sources", "build_vector_store"] diff --git a/harness/tht/adapters/vector/__init__.py b/harness/tht/adapters/vector/__init__.py index 4a3421cb..57ac94a7 100644 --- a/harness/tht/adapters/vector/__init__.py +++ b/harness/tht/adapters/vector/__init__.py @@ -1,8 +1,5 @@ """Vector-store adapter implementations.""" -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", "QdrantVectorStore", "ThothHttpVectorStore"] +__all__ = ["QdrantVectorStore"] diff --git a/harness/tht/adapters/vector/_shared.py b/harness/tht/adapters/vector/_shared.py new file mode 100644 index 00000000..fd1df549 --- /dev/null +++ b/harness/tht/adapters/vector/_shared.py @@ -0,0 +1,41 @@ +"""Shared collection and kind validation for vector stores.""" + +from __future__ import annotations + +from tht.ports.vector import VectorStoreError + +COLLECTION_KINDS = { + "schema_records": {"schema_table", "schema_column"}, + "evidence": {"evidence"}, + "memory": {"memory", "solved_question"}, +} +ALLOWED_COLLECTIONS = frozenset(COLLECTION_KINDS) +ALLOWED_KINDS = frozenset().union(*COLLECTION_KINDS.values()) + + +def validate_collection(collection: str) -> str: + if collection not in ALLOWED_COLLECTIONS: + raise VectorStoreError(f"Collection not allowed: {collection}") + return collection + + +def validate_collection_kinds(collection: str, kinds: list[str]) -> None: + invalid = set(kinds) - COLLECTION_KINDS[collection] + if invalid: + raise VectorStoreError(f"Kind not allowed for {collection}: {', '.join(sorted(invalid))}") + + +def validate_known_kinds(kinds: list[str]) -> None: + invalid = set(kinds) - ALLOWED_KINDS + if invalid: + raise VectorStoreError(f"Kind not allowed: {', '.join(sorted(invalid))}") + + +__all__ = [ + "ALLOWED_COLLECTIONS", + "ALLOWED_KINDS", + "COLLECTION_KINDS", + "validate_collection", + "validate_collection_kinds", + "validate_known_kinds", +] diff --git a/harness/tht/adapters/vector/legacy_direct.py b/harness/tht/adapters/vector/legacy_direct.py deleted file mode 100644 index c14202f3..00000000 --- a/harness/tht/adapters/vector/legacy_direct.py +++ /dev/null @@ -1,70 +0,0 @@ -"""Compatibility adapter for the existing direct PostgreSQL vector reader.""" - -from sqlalchemy import Engine - -from tht.ports.vector import ( - VectorCapabilities, - VectorHealth, - VectorStoreError, - VectorWriteRecord, - VectorWriteUnavailable, - require_positive_limit, -) -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), - read_configured=True, - read_reachable=False, - read_detail=str(exc), - expected_dimension=self._dim, - ) - return VectorHealth( - ok=True, - read_configured=True, - read_reachable=True, - expected_dimension=self._dim, - ) - - 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) - if metadata_filter is not None: - raise VectorStoreError("Legacy vector store cannot enforce metadata filtering") - 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[VectorWriteRecord]) -> int: - raise VectorWriteUnavailable("Legacy direct reader has no writer interface") diff --git a/harness/tht/adapters/vector/pgvector.py b/harness/tht/adapters/vector/pgvector.py deleted file mode 100644 index b4ed96d8..00000000 --- a/harness/tht/adapters/vector/pgvector.py +++ /dev/null @@ -1,524 +0,0 @@ -"""Direct PostgreSQL/pgvector implementation of the vector port.""" - -import json -import re - -from psycopg2 import Error as PsycopgError -from psycopg2 import sql -from sqlalchemy import Engine -from sqlalchemy.exc import SQLAlchemyError - -from tht.config import DatabaseConfig -from tht.db.connection import make_engine -from tht.ports.vector import ( - VectorCapabilities, - VectorHealth, - VectorReadUnavailable, - VectorStoreError, - VectorWriteRecord, - VectorWriteUnavailable, - require_positive_limit, -) -from tht.vectorstore.store import VectorHit, hit_from_metadata - -COLLECTION_KINDS = { - "schema_records": {"schema_table", "schema_column"}, - "evidence": {"evidence"}, - "memory": {"memory", "solved_question"}, -} -ALLOWED_COLLECTIONS = frozenset(COLLECTION_KINDS) -ALLOWED_KINDS = frozenset().union(*COLLECTION_KINDS.values()) -_VECTOR_DIMENSION = re.compile(r"^(?:[a-z_][a-z0-9_]*\.)?vector\((\d+)\)$") - - -def _collection(schema: str, name: str) -> sql.Identifier: - if name not in ALLOWED_COLLECTIONS: - raise VectorStoreError(f"Collection not allowed: {name}") - return sql.Identifier(schema, name) - - -def _vector_literal(values: list[float]) -> str: - return "[" + ",".join(str(float(value)) for value in values) + "]" - - -def _vector_type(schema: str) -> sql.Identifier: - return sql.Identifier(schema, "vector") - - -def _cosine_operator(schema: str) -> sql.Composed: - return sql.SQL("OPERATOR({}.<=>)").format(sql.Identifier(schema)) - - -def _vector_sql_names(cursor, table_schema: str, collection: str) -> tuple[str, str]: - """Discover pgvector type and operator namespaces from the embedding column.""" - cursor.execute( - """SELECT type_ns.nspname, operator_ns.nspname - FROM pg_catalog.pg_attribute attribute - JOIN pg_catalog.pg_class table_class - ON table_class.oid = attribute.attrelid - JOIN pg_catalog.pg_namespace table_ns - ON table_ns.oid = table_class.relnamespace - JOIN pg_catalog.pg_type vector_type - ON vector_type.oid = attribute.atttypid - JOIN pg_catalog.pg_namespace type_ns - ON type_ns.oid = vector_type.typnamespace - JOIN pg_catalog.pg_operator cosine - ON cosine.oprname = %s - AND cosine.oprleft = vector_type.oid - AND cosine.oprright = vector_type.oid - JOIN pg_catalog.pg_namespace operator_ns - ON operator_ns.oid = cosine.oprnamespace - WHERE table_ns.nspname = %s - AND table_class.relname = %s - AND attribute.attname = %s - AND NOT attribute.attisdropped - ORDER BY cosine.oid - LIMIT 1""", - ("<=>", table_schema, collection, "embedding"), - ) - row = cursor.fetchone() - if row is None: - raise VectorStoreError(f"Collection {collection} has no usable pgvector embedding") - return row[0], row[1] - - -def _validate_collection_kinds(collection: str, kinds: list[str]) -> None: - invalid = set(kinds) - COLLECTION_KINDS[collection] - if invalid: - raise VectorStoreError(f"Kind not allowed for {collection}: {', '.join(sorted(invalid))}") - - -def _validate_known_kinds(kinds: list[str]) -> None: - invalid = set(kinds) - ALLOWED_KINDS - if invalid: - raise VectorStoreError(f"Kind not allowed: {', '.join(sorted(invalid))}") - - -class PgVectorStore: - """Direct store with independent reader and writer database credentials.""" - - def __init__( - self, - read_config: DatabaseConfig | None, - write_config: DatabaseConfig | None = None, - *, - expected_dimension: int | None = None, - ): - self._reader = make_engine(read_config) if read_config is not None else None - self._writer = make_engine(write_config) if write_config is not None else None - config = read_config or write_config - self._schema = config.db_schema if config is not None else "vectors" - if read_config and write_config and read_config.db_schema != write_config.db_schema: - raise VectorStoreError("Reader and writer vector schemas must match") - self._expected_dimension = expected_dimension - - @property - def capabilities(self) -> VectorCapabilities: - writable = self._writer is not None - return VectorCapabilities( - search=self._reader is not None, - existing_hashes=writable, - upsert=writable, - metadata_filter=self._reader is not None, - delete_generation=writable, - list_evidence_generations=writable, - ) - - def _probe( - self, engine: Engine | None, *, writable: bool - ) -> tuple[bool | None, str | None, set[int]]: - if engine is None: - return None, None, set() - try: - raw = engine.raw_connection() - try: - with raw.cursor() as cursor: - cursor.execute("SELECT 1") - cursor.execute( - "SELECT has_schema_privilege(current_user, %s, 'USAGE')", - (self._schema,), - ) - schema_usage = bool(cursor.fetchone()[0]) - if not schema_usage: - return False, "vector schema incomplete: missing schema usage", set() - cursor.execute( - """SELECT c.relname, format_type(a.atttypid, a.atttypmod), - has_table_privilege(current_user, c.oid, 'SELECT'), - has_table_privilege(current_user, c.oid, 'INSERT'), - has_table_privilege(current_user, c.oid, 'UPDATE'), - has_column_privilege(current_user, c.oid, 'record_key', 'SELECT') - AND has_column_privilege( - current_user, c.oid, 'content_hash', 'SELECT' - ) - AND has_column_privilege(current_user, c.oid, 'kind', 'SELECT'), - CASE WHEN id_attr.attname IS NOT NULL THEN - pg_get_serial_sequence( - format('%%I.%%I', n.nspname, c.relname), 'id' - ) - END AS id_sequence, - CASE WHEN id_attr.attname IS NOT NULL THEN - has_sequence_privilege( - current_user, - pg_get_serial_sequence( - format('%%I.%%I', n.nspname, c.relname), 'id' - ), - 'USAGE' - ) - END AS sequence_usage - FROM pg_class c - JOIN pg_namespace n ON n.oid = c.relnamespace - LEFT JOIN pg_attribute a ON a.attrelid = c.oid - AND a.attname = 'embedding' AND NOT a.attisdropped - LEFT JOIN pg_attribute id_attr ON id_attr.attrelid = c.oid - AND id_attr.attname = 'id' AND NOT id_attr.attisdropped - WHERE n.nspname = %s AND c.relname = ANY(%s) - AND c.relkind IN ('r', 'p')""", - (self._schema, list(ALLOWED_COLLECTIONS)), - ) - rows = cursor.fetchall() - present = {row[0] for row in rows} - missing_tables = sorted(ALLOWED_COLLECTIONS - present) - missing_embeddings = sorted(row[0] for row in rows if row[1] is None) - privilege_missing = sorted( - row[0] - for row in rows - if (writable and not (row[3] and row[4] and row[5])) - or (not writable and not row[2]) - ) - missing_sequences = sorted( - row[0] for row in rows if writable and row[6] is None - ) - sequence_privilege_missing = sorted( - row[0] for row in rows if writable and row[6] is not None and not row[7] - ) - problems = [] - if missing_tables: - problems.append("missing tables " + ", ".join(missing_tables)) - if missing_embeddings: - problems.append( - "missing embedding columns " + ", ".join(missing_embeddings) - ) - if privilege_missing: - authority = "write" if writable else "read" - problems.append( - f"missing {authority} privileges " + ", ".join(privilege_missing) - ) - if missing_sequences: - problems.append("missing id sequences " + ", ".join(missing_sequences)) - if sequence_privilege_missing: - problems.append( - "missing sequence privileges " + ", ".join(sequence_privilege_missing) - ) - if problems: - return False, "vector schema incomplete: " + "; ".join(problems), set() - dimensions = { - int(match.group(1)) - for _, type_name, *_ in rows - if (match := _VECTOR_DIMENSION.match(type_name)) - } - invalid_types = sorted( - row[0] - for row in rows - if row[1] is not None and not _VECTOR_DIMENSION.match(row[1]) - ) - if invalid_types: - return ( - False, - "vector schema incomplete: invalid embedding types " - + ", ".join(invalid_types), - set(), - ) - if self._expected_dimension is not None: - mismatches = sorted( - f"{name}={int(match.group(1))}" - for name, type_name, *_ in rows - if (match := _VECTOR_DIMENSION.match(type_name)) - and int(match.group(1)) != self._expected_dimension - ) - if mismatches: - return ( - False, - "embedding dimension mismatch: " + ", ".join(mismatches), - dimensions, - ) - return True, None, dimensions - finally: - raw.close() - except (AttributeError, TypeError, ValueError, PsycopgError, SQLAlchemyError) as exc: - return False, f"vector database probe failed: {type(exc).__name__}", set() - - def health(self) -> VectorHealth: - read_ok, read_detail, read_dimensions = self._probe(self._reader, writable=False) - write_ok, write_detail, write_dimensions = self._probe(self._writer, writable=True) - dimensions = tuple(sorted(read_dimensions | write_dimensions)) - compatible = ( - None - if self._expected_dimension is None or not dimensions - else dimensions == (self._expected_dimension,) - ) - reachable = [value for value in (read_ok, write_ok) if value is not None] - details = [value for value in (read_detail, write_detail) if value] - return VectorHealth( - ok=bool(reachable) and all(reachable) and compatible is not False, - detail="; ".join(details) or None, - read_configured=self._reader is not None, - read_reachable=read_ok, - read_detail=read_detail, - write_configured=self._writer is not None, - write_reachable=write_ok, - write_detail=write_detail, - 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) - if self._reader is None: - raise VectorReadUnavailable("Vector reader credential is not configured") - if self._expected_dimension is not None and len(embedding) != self._expected_dimension: - raise VectorStoreError("Query embedding dimension does not match configured dimension") - if kinds: - _validate_known_kinds(kinds) - hits: list[VectorHit] = [] - raw = None - try: - raw = self._reader.raw_connection() - with raw.cursor() as cursor: - for collection in collections: - table = _collection(self._schema, collection) - type_schema, operator_schema = _vector_sql_names( - cursor, self._schema, collection - ) - collection_kinds = ( - sorted(set(kinds) & COLLECTION_KINDS[collection]) if kinds else None - ) - if kinds and not collection_kinds: - continue - clauses = [] - filter_params = [] - if collection_kinds: - clauses.append(sql.SQL("kind = ANY(%s)")) - filter_params.append(collection_kinds) - if metadata_filter is not None: - if collection != "evidence" or 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 isinstance(workspace_id, str): - raise VectorStoreError("Invalid vector metadata filter") - clauses.append(sql.SQL("metadata->>'vector_generation' = %s")) - clauses.append(sql.SQL("metadata->>'document_id' = ANY(%s)")) - clauses.append(sql.SQL("metadata->>'workspace_id' = %s")) - filter_params.extend((generation, document_ids, workspace_id)) - where = ( - sql.SQL(" WHERE ") + sql.SQL(" AND ").join(clauses) - if clauses else sql.SQL("") - ) - query = sql.SQL( - "SELECT metadata, 1 - (embedding {} %s::{}) AS similarity " - "FROM {}{} ORDER BY embedding {} %s::{}, record_key LIMIT %s" - ).format( - _cosine_operator(operator_schema), - _vector_type(type_schema), - table, - where, - _cosine_operator(operator_schema), - _vector_type(type_schema), - ) - params = [_vector_literal(embedding)] - params.extend(filter_params) - params.extend((_vector_literal(embedding), limit)) - cursor.execute(query, params) - hits.extend(hit_from_metadata(row[1], row[0]) for row in cursor.fetchall()) - except VectorStoreError: - raise - except Exception as exc: - raise VectorReadUnavailable("Vector read operation unavailable") from exc - finally: - if raw is not None: - raw.close() - return sorted(hits, key=lambda hit: (-hit.similarity, hit.id))[:limit] - - def _require_writer(self) -> Engine: - 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]: - engine = self._require_writer() - table = _collection(self._schema, collection) - _validate_collection_kinds(collection, kinds) - raw = None - try: - raw = engine.raw_connection() - with raw.cursor() as cursor: - cursor.execute( - sql.SQL("SELECT record_key, content_hash FROM {} WHERE kind = ANY(%s)").format( - table - ), - (kinds,), - ) - return dict(cursor.fetchall()) - except VectorStoreError: - raise - except Exception as exc: - raise VectorWriteUnavailable("Vector write operation unavailable") from exc - finally: - if raw is not None: - raw.close() - - def upsert(self, collection: str, records: list[VectorWriteRecord]) -> int: - engine = self._require_writer() - table = _collection(self._schema, collection) - for write_record in records: - _validate_collection_kinds(collection, [write_record.record.kind]) - if ( - self._expected_dimension is not None - and len(write_record.embedding) != self._expected_dimension - ): - raise VectorStoreError("Embedding dimension does not match configured dimension") - raw = None - try: - raw = engine.raw_connection() - with raw.cursor() as cursor: - type_schema, _ = _vector_sql_names(cursor, self._schema, collection) - insert = sql.SQL( - "INSERT INTO {} (record_key, kind, content_hash, metadata, embedding) " - "VALUES (%s, %s, %s, %s::jsonb, %s::{}) " - "ON CONFLICT (record_key) DO NOTHING" - ).format(table, _vector_type(type_schema)) - update = sql.SQL( - "UPDATE {} SET kind = %s, content_hash = %s, metadata = %s::jsonb, " - "embedding = %s::{}, indexed_at = pg_catalog.now() WHERE record_key = %s" - ).format(table, _vector_type(type_schema)) - for write_record in records: - record = write_record.record - metadata = { - "kind": record.kind, - "ref": record.ref, - "record_key": record.id, - "title": record.title, - "content": record.content, - **record.metadata, - } - metadata_json = json.dumps(metadata) - vector = _vector_literal(write_record.embedding) - cursor.execute( - insert, - (record.id, record.kind, write_record.content_hash, metadata_json, vector), - ) - if cursor.rowcount == 0: - cursor.execute( - update, - ( - record.kind, - write_record.content_hash, - metadata_json, - vector, - record.id, - ), - ) - raw.commit() - except VectorStoreError: - if raw is not None: - raw.rollback() - raise - except Exception as exc: - if raw is not None: - raw.rollback() - raise VectorWriteUnavailable("Vector write operation unavailable") from exc - finally: - if raw is not None: - raw.close() - return len(records) - - def delete_generation(self, collection: str, generation: str, workspace_id: str) -> int: - if collection != "evidence" or re.fullmatch(r"gen:[0-9a-f]{32}", generation) is None: - raise VectorStoreError("Only exact Evidence generations may be deleted") - if re.fullmatch(r"[a-z][a-z0-9_-]{0,63}", workspace_id) is None: - raise VectorStoreError("Invalid Evidence workspace namespace") - raw = None - try: - raw = self._require_writer().raw_connection() - with raw.cursor() as cursor: - cursor.execute( - sql.SQL( - "DELETE FROM {} WHERE kind = 'evidence' " - "AND metadata->>'vector_generation' = %s " - "AND metadata->>'workspace_id' = %s" - ).format(_collection(self._schema, collection)), - (generation, workspace_id), - ) - count = cursor.rowcount - raw.commit() - return count - except Exception as exc: - if raw is not None: - raw.rollback() - raise VectorWriteUnavailable("Vector generation cleanup unavailable") from exc - finally: - if raw is not None: - raw.close() - - def delete_kinds(self, collection: str, kinds: list[str]) -> int: - _collection(self._schema, collection) - _validate_collection_kinds(collection, kinds) - raw = None - try: - raw = self._require_writer().raw_connection() - with raw.cursor() as cursor: - cursor.execute( - sql.SQL("DELETE FROM {} WHERE kind = ANY(%s)").format( - _collection(self._schema, collection) - ), - (kinds,), - ) - count = cursor.rowcount - raw.commit() - return count - except Exception as exc: - if raw is not None: - raw.rollback() - raise VectorWriteUnavailable("Vector kind cleanup unavailable") from exc - finally: - if raw is not None: - raw.close() - - 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 re.fullmatch(r"[a-z][a-z0-9_-]{0,63}", workspace_id) is None: - raise VectorStoreError("Invalid Evidence workspace namespace") - raw = None - try: - raw = self._require_writer().raw_connection() - with raw.cursor() as cursor: - cursor.execute( - sql.SQL( - "SELECT DISTINCT metadata->>'vector_generation' FROM {} " - "WHERE kind = 'evidence' AND metadata->>'vector_generation' " - "~ '^gen:[0-9a-f]{{32}}$' AND metadata->>'workspace_id' = %s ORDER BY 1" - ).format(_collection(self._schema, collection)), - (workspace_id,), - ) - return [row[0] for row in cursor.fetchall()] - except Exception as exc: - raise VectorWriteUnavailable("Vector generation inventory unavailable") from exc - finally: - if raw is not None: - raw.close() - - -__all__ = ["ALLOWED_COLLECTIONS", "PgVectorStore"] diff --git a/harness/tht/adapters/vector/qdrant.py b/harness/tht/adapters/vector/qdrant.py index f8804eeb..7d880d82 100644 --- a/harness/tht/adapters/vector/qdrant.py +++ b/harness/tht/adapters/vector/qdrant.py @@ -6,11 +6,11 @@ from uuid import NAMESPACE_URL, uuid5 import requests -from tht.adapters.vector.pgvector import ( +from tht.adapters.vector._shared import ( COLLECTION_KINDS, - _collection, - _validate_collection_kinds, - _validate_known_kinds, + validate_collection, + validate_collection_kinds, + validate_known_kinds, ) from tht.ports.vector import ( VectorCapabilities, @@ -165,8 +165,8 @@ class QdrantVectorStore: 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) + validate_collection(collection) + validate_collection_kinds(collection, kinds) points = self._scroll( [ *self._workspace_filter(), @@ -186,11 +186,11 @@ class QdrantVectorStore: return hashes def upsert(self, collection: str, records: list[VectorWriteRecord]) -> int: - _collection("vectors", collection) + validate_collection(collection) self._ensure_collection(strict=True) points = [] for write_record in records: - _validate_collection_kinds(collection, [write_record.record.kind]) + 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( @@ -213,8 +213,8 @@ class QdrantVectorStore: return len(records) def delete_kinds(self, collection: str, kinds: list[str]) -> int: - _collection("vectors", collection) - _validate_collection_kinds(collection, kinds) + validate_collection(collection) + validate_collection_kinds(collection, kinds) must = [ *self._workspace_filter(), {"key": "record_kind", "match": {"any": sorted(kinds)}}, @@ -284,10 +284,10 @@ class QdrantVectorStore: ) -> list[str]: selected: set[str] = set() for collection in collections: - _collection("vectors", collection) + validate_collection(collection) selected.update(COLLECTION_KINDS[collection]) if kinds: - _validate_known_kinds(kinds) + validate_known_kinds(kinds) selected &= set(kinds) return sorted(selected) diff --git a/harness/tht/adapters/vector/thoth_http.py b/harness/tht/adapters/vector/thoth_http.py deleted file mode 100644 index a202af14..00000000 --- a/harness/tht/adapters/vector/thoth_http.py +++ /dev/null @@ -1,191 +0,0 @@ -"""Thoth vector HTTP adapter using distinct read and write clients.""" - -import re - -from tht.adapters.vector.pgvector import ( - _collection, - _validate_collection_kinds, - _validate_known_kinds, -) -from tht.ports.vector import ( - VectorCapabilities, - VectorHealth, - VectorHit, - VectorReadUnavailable, - VectorStoreError, - VectorWriteRecord, - VectorWriteUnavailable, - require_positive_limit, -) -from tht.vectorstore.rest_client import VectorRestClient, VectorRestError -from tht.vectorstore.store import hit_from_metadata - - -def _merge(hits: list[VectorHit], limit: int) -> list[VectorHit]: - return sorted(hits, key=lambda hit: (-hit.similarity, hit.id))[:limit] - - -class ThothHttpVectorStore: - """Vector port backed by the existing allowlisted REST RPCs.""" - - def __init__( - self, - reader: VectorRestClient | None, - writer: VectorRestClient | None, - expected_dimension: int | None = None, - ): - self._reader = reader - self._writer = writer - self._expected_dimension = expected_dimension - - @property - def capabilities(self) -> VectorCapabilities: - writable = self._writer is not None - return VectorCapabilities( - search=self._reader is not None, existing_hashes=writable, upsert=writable, - metadata_filter=self._reader is not None, delete_generation=writable, - list_evidence_generations=writable, - ) - - def health(self) -> VectorHealth: - read_reachable, read_detail, read_tables = self._probe(self._reader) - write_reachable, write_detail, write_tables = self._probe(self._writer) - dimensions = tuple(sorted({ - dimension - for row in [*read_tables, *write_tables] - if type(dimension := row.get("vector_dimensions")) is int - })) - compatible = ( - None - if self._expected_dimension is None or not dimensions - else dimensions == (self._expected_dimension,) - ) - reachable = [ - status for status in (read_reachable, write_reachable) if status is not None - ] - ok = bool(reachable) and all(reachable) and compatible is not False - details = [detail for detail in (read_detail, write_detail) if detail] - return VectorHealth( - ok=ok, - detail="; ".join(details) or None, - read_configured=self._reader is not None, - read_reachable=read_reachable, - read_detail=read_detail, - write_configured=self._writer is not None, - write_reachable=write_reachable, - write_detail=write_detail, - expected_dimension=self._expected_dimension, - observed_dimensions=dimensions, - dimension_compatible=compatible, - ) - - @staticmethod - def _probe(client: VectorRestClient | None) -> tuple[bool | None, str | None, list[dict]]: - if client is None: - return None, None, [] - try: - return True, None, client.list_tables() - except (RuntimeError, VectorRestError) as exc: - return False, str(exc), [] - - 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) - if self._reader is None: - raise VectorReadUnavailable("Vector reader credential is not configured") - if self._expected_dimension is not None and len(embedding) != self._expected_dimension: - raise VectorStoreError("Query embedding dimension does not match configured dimension") - if kinds: - _validate_known_kinds(kinds) - hits: list[VectorHit] = [] - for collection in collections: - _collection("vectors", collection) - try: - if metadata_filter is None: - rows = self._reader.search_similar(collection, embedding, limit, kinds=kinds) - else: - rows = self._reader.search_similar( - collection, embedding, limit, kinds=kinds, - metadata_filter=metadata_filter, - ) - except VectorRestError as exc: - raise VectorStoreError(str(exc)) from exc - 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]: - _collection("vectors", collection) - _validate_collection_kinds(collection, kinds) - try: - return self._require_writer().existing_hashes(collection, kinds) - except VectorRestError as exc: - raise VectorStoreError(str(exc)) from exc - - def upsert(self, collection: str, records: list[VectorWriteRecord]) -> int: - writer = self._require_writer() - _collection("vectors", collection) - for record in records: - _validate_collection_kinds(collection, [record.record.kind]) - if ( - self._expected_dimension is not None - and len(record.embedding) != self._expected_dimension - ): - raise VectorStoreError("Embedding dimension does not match configured dimension") - rows = [self._row(record) for record in records] - try: - return writer.upsert_records(collection, rows) - except VectorRestError as exc: - raise VectorStoreError(str(exc)) from exc - - def delete_generation(self, collection: str, generation: str, workspace_id: str) -> int: - if collection != "evidence" or re.fullmatch(r"gen:[0-9a-f]{32}", generation) is None: - raise VectorStoreError("Only exact Evidence generations may be deleted") - try: - return self._require_writer().delete_generation(collection, generation, workspace_id) - except VectorRestError as exc: - raise VectorStoreError(str(exc)) from exc - - def list_evidence_generations(self, collection: str, workspace_id: str) -> list[str]: - if collection != "evidence": - raise VectorStoreError("Only exact Evidence generations may be listed") - try: - return self._require_writer().list_evidence_generations(collection, workspace_id) - except VectorRestError as exc: - raise VectorWriteUnavailable("Vector generation inventory unavailable") from exc - - @staticmethod - def _row(write_record: VectorWriteRecord) -> dict: - record = write_record.record - metadata = { - "kind": record.kind, - "ref": record.ref, - "record_key": record.id, - "title": record.title, - "content": record.content, - **record.metadata, - } - return { - "record_key": record.id, - "kind": record.kind, - "content_hash": write_record.content_hash, - "metadata": metadata, - "embedding": write_record.embedding, - } diff --git a/harness/tht/cli/evidence_cmd.py b/harness/tht/cli/evidence_cmd.py index fcd3165a..b09171f7 100644 --- a/harness/tht/cli/evidence_cmd.py +++ b/harness/tht/cli/evidence_cmd.py @@ -2,9 +2,9 @@ from pathlib import Path import typer +from tht.cli._guards import require_vector_write_allowed from tht.cli.config_cmd import CONFIG_OPT from tht.cli.schema_cmd import _load_config_or_exit -from tht.cli._guards import require_vector_write_allowed evidence_app = typer.Typer(help="Generazione e gestione delle evidence") @@ -56,12 +56,13 @@ def extract_cmd(config: Path = CONFIG_OPT) -> None: @evidence_app.command("index") def index_cmd(config: Path = CONFIG_OPT) -> None: - """Embedda e sincronizza su pgvector tutte le evidence presenti in artifacts/.""" + """Embedda e sincronizza nel semantic store tutte le evidence presenti in artifacts/.""" + from tht.adapters.factory import build_vector_store from tht.cli.vector_cmd import ( _print_stats, make_embedder, - open_store, require_vector_cfg, + sync_canonical_records, ) from tht.evidence.model import load_evidence_dir from tht.vectorstore.records import evidence_records @@ -71,6 +72,10 @@ def index_cmd(config: Path = CONFIG_OPT) -> None: require_vector_cfg(cfg) docs = load_evidence_dir(evidence_root(cfg)) records = evidence_records(docs, cfg.vector.max_chunk_chars) - store = open_store(cfg, "evidence") - stats = store.sync(records, make_embedder(cfg.embeddings), kinds={"evidence"}) + stats = sync_canonical_records( + "evidence", + records, + store=build_vector_store(cfg, require_write=True), + embedder=make_embedder(cfg.embeddings), + ) _print_stats(stats) diff --git a/harness/tht/cli/memory_cmd.py b/harness/tht/cli/memory_cmd.py index fc4d74dc..05146121 100644 --- a/harness/tht/cli/memory_cmd.py +++ b/harness/tht/cli/memory_cmd.py @@ -11,7 +11,6 @@ import typer from sqlalchemy.exc import OperationalError, ProgrammingError from tht.cli._guards import ( - has_vector_write_rest, require_server_profile, require_vector_write_allowed, ) @@ -45,15 +44,8 @@ def _resync_memory(cfg): def clear_memory_index(cfg): from tht.adapters.factory import build_vector_store - from tht.cli.vector_cmd import make_embedder, open_store, require_direct_vector_cfg - if cfg.vectors is not None and cfg.vectors.type == "qdrant": - return build_vector_store(cfg, require_write=True).delete_kinds("memory", ["memory"]) - - require_direct_vector_cfg(cfg) - legacy_store = open_store(cfg, "memory") - legacy_store.sync([], make_embedder(cfg.embeddings), kinds={"memory"}) - return 0 + return build_vector_store(cfg, require_write=True).delete_kinds("memory", ["memory"]) @memory_app.command("promote") @@ -451,19 +443,13 @@ def search_cmd( def index_solved_session(cfg, session_id: str) -> int: """Indicizza la coppia domanda->SQL della sessione (kind solved_question). - Solleva RuntimeError se manca la writer key e SolvedIndexError se mancano gli - artefatti: il finalize li degrada a warning, il comando CLI li converte in - errori espliciti.""" + Solleva SolvedIndexError se mancano gli artefatti: il finalize lo degrada a warning, + il comando CLI lo converte in errore esplicito.""" from tht.adapters.factory import build_vector_store from tht.cli.sql_cmd import promoted_tables_for from tht.cli.vector_cmd import make_embedder from tht.solved import build_solved_snapshot, save_solved_question - if not has_vector_write_rest(cfg): - raise RuntimeError( - "vector_write_rest assente: la coppia domanda->SQL si indicizza solo con la " - "writer key configurata nel workspace yaml" - ) store = build_vector_store(cfg, require_write=True) record = build_solved_snapshot(load_snapshot_or_exit(cfg, session_id), promoted_tables_for(cfg, session_id)) return save_solved_question( @@ -518,10 +504,9 @@ def solved_search_cmd( from rich.table import Table from tht.cli.vector_cmd import make_embedder, open_searcher - from tht.ports.vector import VectorReadUnavailable + from tht.ports.vector import VectorReadUnavailable, VectorStoreError from tht.solved import SOLVED_KIND from tht.vectorstore.embeddings import EmbeddingsError - from tht.vectorstore.rest_client import VectorRestError cfg = _load_config_or_exit(config) require_vector_cfg(cfg) @@ -532,7 +517,7 @@ def solved_search_cmd( searcher = open_searcher(cfg) embedder = make_embedder(cfg.embeddings) hits = searcher.search(embedder.embed_query(question), top_n=top, kinds=[SOLVED_KIND]) - except (VectorRestError, VectorReadUnavailable, EmbeddingsError, OperationalError) as e: + except (VectorStoreError, VectorReadUnavailable, EmbeddingsError, OperationalError) as e: typer.secho( f"ATTENZIONE: exemplar non disponibili ({e}). Prosegui senza.", fg=typer.colors.YELLOW, err=True, diff --git a/harness/tht/cli/search_cmd.py b/harness/tht/cli/search_cmd.py index 7f397a55..d84936bb 100644 --- a/harness/tht/cli/search_cmd.py +++ b/harness/tht/cli/search_cmd.py @@ -255,11 +255,10 @@ def pack_cmd( from sqlalchemy.exc import OperationalError from tht.cli.vector_cmd import make_embedder, open_searcher, require_vector_cfg - from tht.ports.vector import VectorReadUnavailable + from tht.ports.vector import VectorReadUnavailable, VectorStoreError from tht.search import combined_search, schema_tables from tht.solved import SOLVED_KIND from tht.vectorstore.embeddings import EmbeddingsError - from tht.vectorstore.rest_client import VectorRestError cfg = _load_config_or_exit(config) from tht.search.evidence import validate_corpus_workspace @@ -273,7 +272,7 @@ def pack_cmd( evidence: list[dict] = [] solved: list[dict] = [] warnings: list[str] = [] - degrade = (VectorRestError, VectorReadUnavailable, EmbeddingsError, OperationalError) + degrade = (VectorStoreError, VectorReadUnavailable, EmbeddingsError, OperationalError) vec = None searcher = embedder = None diff --git a/harness/tht/cli/vector_cmd.py b/harness/tht/cli/vector_cmd.py index 3072160a..cf65a3fe 100644 --- a/harness/tht/cli/vector_cmd.py +++ b/harness/tht/cli/vector_cmd.py @@ -2,11 +2,7 @@ from pathlib import Path import typer -from tht.cli._guards import ( - has_vector_write_rest, - require_server_profile, - require_vector_write_allowed, -) +from tht.cli._guards import require_server_profile, require_vector_write_allowed from tht.cli.config_cmd import CONFIG_OPT from tht.cli.schema_cmd import _load_config_or_exit, annotations_path, physical_path from tht.ports.vector import VectorWriteRecord @@ -26,8 +22,8 @@ def require_vector_cfg(cfg): missing = [] if cfg.embeddings is None: missing.append("embeddings") - if cfg.vectors is None and cfg.vector_db is None and not has_vector_write_rest(cfg): - missing.append("vectors o vector_db o vector_write_rest") + if cfg.vectors is None: + missing.append("vectors") if missing: typer.secho( f"ERRORE: sezioni mancanti nel workspace yaml: {', '.join(missing)}.", @@ -36,27 +32,6 @@ def require_vector_cfg(cfg): raise typer.Exit(code=1) -def require_direct_vector_cfg(cfg): - missing = [k for k in ("vector_db", "embeddings") if getattr(cfg, k) is None] - if missing: - typer.secho( - f"ERRORE: sezioni mancanti nel workspace yaml: {', '.join(missing)}.", - fg=typer.colors.RED, err=True, - ) - raise typer.Exit(code=1) - - -def open_store(cfg, table: str): - """Writer table-scoped per il LOADING. - - Sul server preferisce la connessione diretta. In profilo workstation usa `vector_write_rest` - se configurato, con upsert remoto non distruttivo. - """ - from tht.adapters.factory import build_vector_loader - - return build_vector_loader(cfg, table) - - def open_searcher(cfg): """Searcher per la LETTURA (similarity search): via REST se `vector_rest` è configurato, altrimenti connessione diretta (dev/test).""" @@ -115,20 +90,20 @@ def init_cmd( False, "--skip-ollama-check", help="Non verificare la raggiungibilita' di Ollama." ), ) -> None: - """Crea schema e tabella pgvector (idempotente) e verifica le connessioni.""" - from sqlalchemy.exc import OperationalError - + """Verifica il runtime Qdrant e la raggiungibilita' dell'embedder configurato.""" + from tht.adapters.factory import build_vector_store from tht.vectorstore.embeddings import EmbeddingsError - from tht.vectorstore.reader import ALL_TABLES cfg = _load_config_or_exit(config) require_server_profile(cfg, "vector init") - require_direct_vector_cfg(cfg) - try: - for table in ALL_TABLES: - open_store(cfg, table).init_schema() - except OperationalError as e: - typer.secho(f"ERRORE connessione pgvector: {e.orig}", fg=typer.colors.RED, err=True) + require_vector_cfg(cfg) + health = build_vector_store(cfg, require_write=True).health() + if not health.ok: + typer.secho( + f"ERRORE runtime vettoriale: {health.detail or 'Qdrant non raggiungibile o incompatibile'}", + fg=typer.colors.RED, + err=True, + ) raise typer.Exit(code=1) if not skip_ollama_check: try: @@ -137,8 +112,8 @@ def init_cmd( typer.secho(f"ERRORE: {e}", fg=typer.colors.RED, err=True) raise typer.Exit(code=1) typer.secho( - f"OK: schema {cfg.vector_db.db_schema} pronto (tabelle: {', '.join(ALL_TABLES)}) su " - f"{cfg.vector_db.host}:{cfg.vector_db.port}", fg=typer.colors.GREEN, + f"OK: runtime Qdrant pronto per la collezione {cfg.vectors.collection}", + fg=typer.colors.GREEN, ) diff --git a/harness/tht/migrations/vector/001_extensions.sql b/harness/tht/migrations/vector/001_extensions.sql deleted file mode 100644 index f64f725a..00000000 --- a/harness/tht/migrations/vector/001_extensions.sql +++ /dev/null @@ -1,3 +0,0 @@ -CREATE SCHEMA IF NOT EXISTS vectors; -REVOKE ALL ON SCHEMA vectors FROM PUBLIC; -CREATE EXTENSION IF NOT EXISTS vector WITH SCHEMA vectors; diff --git a/harness/tht/migrations/vector/002_schema_tables.sql b/harness/tht/migrations/vector/002_schema_tables.sql deleted file mode 100644 index 2cffaea8..00000000 --- a/harness/tht/migrations/vector/002_schema_tables.sql +++ /dev/null @@ -1,32 +0,0 @@ -CREATE TABLE IF NOT EXISTS vectors.schema_records ( - id bigserial PRIMARY KEY, - record_key text UNIQUE NOT NULL, - kind text NOT NULL, - content_hash text NOT NULL, - metadata jsonb NOT NULL, - embedding vectors.vector(768) NOT NULL, - indexed_at timestamptz NOT NULL DEFAULT pg_catalog.now() -); - -CREATE TABLE IF NOT EXISTS vectors.evidence ( - id bigserial PRIMARY KEY, - record_key text UNIQUE NOT NULL, - kind text NOT NULL, - content_hash text NOT NULL, - metadata jsonb NOT NULL, - embedding vectors.vector(768) NOT NULL, - indexed_at timestamptz NOT NULL DEFAULT pg_catalog.now() -); - -CREATE TABLE IF NOT EXISTS vectors.memory ( - id bigserial PRIMARY KEY, - record_key text UNIQUE NOT NULL, - kind text NOT NULL, - content_hash text NOT NULL, - metadata jsonb NOT NULL, - embedding vectors.vector(768) NOT NULL, - indexed_at timestamptz NOT NULL DEFAULT pg_catalog.now() -); - -REVOKE ALL ON ALL TABLES IN SCHEMA vectors FROM PUBLIC; -REVOKE ALL ON ALL SEQUENCES IN SCHEMA vectors FROM PUBLIC; diff --git a/harness/tht/migrations/vector/003_roles.sql b/harness/tht/migrations/vector/003_roles.sql deleted file mode 100644 index ba88bd2b..00000000 --- a/harness/tht/migrations/vector/003_roles.sql +++ /dev/null @@ -1,23 +0,0 @@ -DO $roles$ -BEGIN - IF NOT EXISTS (SELECT 1 FROM pg_catalog.pg_roles WHERE rolname = 'vector_reader') THEN - CREATE ROLE vector_reader NOLOGIN; - END IF; - IF NOT EXISTS (SELECT 1 FROM pg_catalog.pg_roles WHERE rolname = 'vector_writer') THEN - CREATE ROLE vector_writer NOLOGIN; - END IF; -END -$roles$; - -REVOKE ALL ON SCHEMA vectors FROM vector_reader, vector_writer; -REVOKE ALL ON ALL TABLES IN SCHEMA vectors FROM vector_reader, vector_writer; -REVOKE ALL ON ALL SEQUENCES IN SCHEMA vectors FROM vector_reader, vector_writer; - -GRANT USAGE ON SCHEMA vectors TO vector_reader, vector_writer; -GRANT SELECT ON ALL TABLES IN SCHEMA vectors TO vector_reader; - -GRANT INSERT, UPDATE -ON vectors.schema_records, vectors.evidence, vectors.memory TO vector_writer; -GRANT SELECT (record_key, kind, content_hash) -ON vectors.schema_records, vectors.evidence, vectors.memory TO vector_writer; -GRANT USAGE ON ALL SEQUENCES IN SCHEMA vectors TO vector_writer; diff --git a/harness/tht/migrations/vector/004_evidence_generation_gc.sql b/harness/tht/migrations/vector/004_evidence_generation_gc.sql deleted file mode 100644 index 94d4a4bb..00000000 --- a/harness/tht/migrations/vector/004_evidence_generation_gc.sql +++ /dev/null @@ -1,3 +0,0 @@ --- The writer owns derived-generation reconciliation but not runtime similarity reads. -GRANT SELECT (metadata) ON vectors.evidence TO vector_writer; -GRANT DELETE ON vectors.evidence TO vector_writer; diff --git a/harness/tht/solved.py b/harness/tht/solved.py index 4bff7521..dae242bf 100644 --- a/harness/tht/solved.py +++ b/harness/tht/solved.py @@ -46,7 +46,7 @@ def _solved_hash(record: VectorRecord) -> str: def save_solved_question(record: VectorRecord, *, store, embedder) -> int: """Upsert one-row della coppia domanda->SQL via writer key (stesso pattern di save_one_memory, spec D11): hash dedup client-side, embedding solo se domanda - o SQL sono cambiati. `writer` e' un VectorRestClient (writer key). Ritorna il + o SQL sono cambiati. Ritorna il numero di righe upsertate (0 = invariata).""" from tht.ports.vector import VectorWriteRecord diff --git a/harness/tht/vectorstore/reader.py b/harness/tht/vectorstore/reader.py index b0cad559..dae8e318 100644 --- a/harness/tht/vectorstore/reader.py +++ b/harness/tht/vectorstore/reader.py @@ -1,18 +1,4 @@ -"""Lettura del pgvector dietro un'unica interfaccia `.search(query_vec, top_n, kinds)`, così -`search.combined_search` resta agnostico al transport. Due implementazioni: - -- `RestSearcher` → produzione: similarity search via REST (`search_similar`). -- `DirectSearcher` → dev/test: connessione diretta a Postgres/pgvector. - -Entrambe mappano i `kind` sulle tabelle per-dominio dello schema `vectors`. -""" - -from sqlalchemy import Engine - -from tht.adapters.vector.legacy_direct import LegacyDirectVectorStore -from tht.adapters.vector.thoth_http import ThothHttpVectorStore -from tht.vectorstore.rest_client import VectorRestClient -from tht.vectorstore.store import VectorHit +"""Collection mapping helpers for the workspace semantic store.""" # kind Thoth → tabella dello schema `vectors`. KIND_TO_TABLE = { @@ -30,35 +16,4 @@ def tables_for_kinds(kinds: list[str] | None) -> list[str]: if not kinds: return list(ALL_TABLES) return sorted({KIND_TO_TABLE[k] for k in kinds if k in KIND_TO_TABLE}) - - -class RestSearcher: - """Similarity search via REST: una chiamata `search_similar` per tabella, poi fusione.""" - - def __init__(self, client: VectorRestClient): - self.client = client - self._store = ThothHttpVectorStore(reader=client, writer=None) - - def search( - self, query_vec: list[float], top_n: int = 10, kinds: list[str] | None = None - ) -> list[VectorHit]: - return self._store.search( - tables_for_kinds(kinds), query_vec, limit=top_n, kinds=kinds - ) - - -class DirectSearcher: - """Similarity search diretta su Postgres/pgvector, interrogando le tabelle per-dominio.""" - - def __init__(self, engine: Engine, schema: str = "vectors", dim: int = 768): - self.engine = engine - self.schema = schema - self.dim = dim - self._store = LegacyDirectVectorStore(engine, schema=schema, dim=dim) - - def search( - self, query_vec: list[float], top_n: int = 10, kinds: list[str] | None = None - ) -> list[VectorHit]: - return self._store.search( - tables_for_kinds(kinds), query_vec, limit=top_n, kinds=kinds - ) +__all__ = ["ALL_TABLES", "KIND_TO_TABLE", "tables_for_kinds"] diff --git a/harness/tht/vectorstore/rest_client.py b/harness/tht/vectorstore/rest_client.py deleted file mode 100644 index b94dab79..00000000 --- a/harness/tht/vectorstore/rest_client.py +++ /dev/null @@ -1,173 +0,0 @@ -"""Client per la similarity search del pgvector esposta via Supabase/PostgREST. - -Endpoint dedicato (es. https://host/vector/v1/), distinto dal DWH. La lettura usa -`search_similar`; la scrittura remota usa RPC allowlist con una API key separata. -Errori in italiano e azionabili, stile `rest/client.py`. -""" - -import re - -import requests - -from tht.config import RestConfig - - -class VectorRestError(Exception): - """Errore di accesso al vector store via REST, con messaggio leggibile per il reviewer.""" - - -class VectorRestClient: - def __init__(self, cfg: RestConfig): - self.cfg = cfg - self._base = cfg.base_url.rstrip("/") - - @property - def api_key(self) -> str: - """The REST API key for this client (spec D11: reader and writer carry - distinct keys against the same endpoint).""" - return self.cfg.api_key - - def _post(self, fn: str, args: dict) -> requests.Response: - url = f"{self._base}/rpc/{fn}" - verify: bool | str = self.cfg.ssl_ca if self.cfg.ssl_ca else True - try: - return requests.post( - url, - json=args, - headers={"X-API-Key": self.cfg.api_key}, - timeout=(self.cfg.connect_timeout, self.cfg.timeout), - verify=verify, - ) - except requests.RequestException as e: - raise VectorRestError( - f"Vector REST non raggiungibile su {self.cfg.base_url} (rpc {fn}): {e}" - ) from e - - def _error_msg(self, fn: str, resp: requests.Response) -> str: - try: - body = resp.json() - detail = body.get("message") or body.get("details") or resp.text - except ValueError: - detail = resp.text - return f"Vector REST rpc {fn} → HTTP {resp.status_code}: {detail}" - - def _call(self, fn: str, args: dict): - resp = self._post(fn, args) - if not resp.ok: - raise VectorRestError(self._error_msg(fn, resp)) - if resp.status_code == 204 or not resp.text: - return None - return resp.json() - - def search_similar( - self, table_name: str, query_embedding: list[float], limit_count: int, - kinds: list[str] | None = None, - metadata_filter: dict | None = None, - ) -> list[dict]: - """Ricerca per similarità coseno su `vectors.`: ritorna le righe - `{id, similarity, metadata}` ordinate per similarity decrescente. Con `kinds` - il filtro avviene server-side nel WHERE della RPC (evita la diluizione del - top-k quando piu' kind condividono la tabella, es. memory/solved_question). - Su un server legacy senza il parametro (PostgREST 404) ritenta senza filtro: - resta il post-filter client-side di RestSearcher.""" - args = { - "query_embedding": query_embedding, - "limit_count": limit_count, - "table_name": table_name, - } - if metadata_filter is not None: - # ACTIVE corpus reads must never degrade to an unfiltered legacy RPC: - # filtering after LIMIT is incomplete and could expose stale generations. - return self._call( - "search_similar", - {**args, "kinds": kinds, "metadata_filter": metadata_filter}, - ) or [] - if kinds is not None: - try: - return self._call("search_similar", {**args, "kinds": kinds}) or [] - except VectorRestError as e: - if "HTTP 404" not in str(e): - raise - # funzione a 3 argomenti (pre-migrazione kinds): fallback senza filtro - return self._call("search_similar", args) or [] - - def list_tables(self) -> list[dict]: - """Tabelle vettoriali disponibili: `{table_name, vector_dimensions, …}`.""" - return self._call("list_tables", {}) or [] - - def existing_hashes(self, table_name: str, kinds: list[str]) -> dict[str, str]: - """Hash correnti per sync incrementale su una tabella vector allowlisted. - - RPC attesa: `existing_vector_hashes(table_name, kinds)` -> righe - `{record_key, content_hash}`. - """ - rows = self._call( - "existing_vector_hashes", - {"table_name": table_name, "kinds": kinds}, - ) or [] - return {row["record_key"]: row["content_hash"] for row in rows} - - def upsert_records(self, table_name: str, rows: list[dict]) -> int: - """Upsert controllato di record vettoriali già embeddati. - - RPC attesa: `upsert_vector_records(table_name, rows)` -> `{upserted: N}` o righe. - Non espone delete/clear: il cleanup distruttivo resta solo-server. - """ - payload = self._call( - "upsert_vector_records", - {"table_name": table_name, "rows": rows}, - ) - if payload is None: - return len(rows) - if isinstance(payload, dict): - return int(payload.get("upserted", len(rows))) - # PostgREST puo' incapsulare uno scalar jsonb in una lista [{"upserted": N}]: - # estrai il conteggio dal primo elemento invece di restituire len(lista)=1. - if isinstance(payload, list): - if payload and isinstance(payload[0], dict) and "upserted" in payload[0]: - return int(payload[0]["upserted"]) - return len(payload) - return len(rows) - - def delete_generation(self, table_name: str, generation: str, workspace_id: str) -> int: - if table_name != "evidence" or re.fullmatch(r"gen:[0-9a-f]{32}", generation) is None: - raise ValueError("generation must be canonical") - if re.fullmatch(r"[a-z][a-z0-9_-]{0,63}", workspace_id) is None: - raise ValueError("workspace namespace must be canonical") - try: - payload = self._call( - "delete_vector_generation", - {"table_name": table_name, "kind": "evidence", "generation": generation, - "workspace_id": workspace_id}, - ) - except VectorRestError as error: - if "HTTP 404" in str(error): - raise VectorRestError( - "delete_vector_generation RPC is unavailable; deploy the cleanup migration" - ) from None - raise - if isinstance(payload, dict): - return int(payload.get("deleted", 0)) - return 0 - - def list_evidence_generations(self, table_name: str, workspace_id: str) -> list[str]: - if re.fullmatch(r"[a-z][a-z0-9_-]{0,63}", workspace_id) is None: - raise ValueError("workspace namespace must be canonical") - try: - rows = self._call( - "list_evidence_generations", - {"table_name": table_name, "kind": "evidence", "workspace_id": workspace_id}, - ) or [] - except VectorRestError as error: - if "HTTP 404" in str(error): - raise VectorRestError( - "list_evidence_generations RPC is unavailable; deploy the cleanup migration" - ) from None - raise - if not isinstance(rows, list) or any( - not isinstance(row, dict) - or re.fullmatch(r"gen:[0-9a-f]{32}", str(row.get("generation", ""))) is None - for row in rows - ): - raise VectorRestError("list_evidence_generations returned malformed data") - return sorted({row["generation"] for row in rows}) diff --git a/harness/tht/vectorstore/rest_writer.py b/harness/tht/vectorstore/rest_writer.py deleted file mode 100644 index b0a4d6d7..00000000 --- a/harness/tht/vectorstore/rest_writer.py +++ /dev/null @@ -1,84 +0,0 @@ -"""Scrittura controllata del pgvector via REST. - -Usata dalle postazioni remote solo quando e' configurata una seconda API key di scrittura. -Mantiene l'upsert incrementale del VectorStore diretto, ma non esegue delete/clear: le -operazioni distruttive restano solo-server via connessione Postgres diretta. -""" - -from tht.vectorstore.records import VectorRecord -from tht.vectorstore.rest_client import VectorRestClient -from tht.vectorstore.store import SyncStats, content_hash - - -TABLE_TO_KINDS = { - "schema_records": {"schema_table", "schema_column"}, - "evidence": {"evidence"}, - "memory": {"memory", "solved_question"}, -} - - -def pack_metadata(record: VectorRecord) -> dict: - """Impacchetta nel metadata tutta la semantica letta poi da `search_similar`.""" - return { - "kind": record.kind, - "ref": record.ref, - "record_key": record.id, - "title": record.title, - "content": record.content, - **record.metadata, - } - - -class RestVectorWriter: - """Writer table-scoped via RPC REST allowlist. - - Il metodo `sync` e' volutamente upsert-only: aggiorna/aggiunge record, conta gli stale, - ma non li elimina. Per cleanup completo usare i comandi server-side con `vector_db`. - """ - - def __init__(self, client: VectorRestClient, table: str): - if table not in TABLE_TO_KINDS: - raise ValueError(f"Tabella vector non supportata per scrittura REST: {table}") - self.client = client - self.table = table - - def existing_hashes(self, kinds: set[str]) -> dict[str, str]: - allowed = TABLE_TO_KINDS[self.table] - bad = kinds - allowed - if bad: - raise ValueError( - f"Kind non ammessi per vectors.{self.table}: {', '.join(sorted(bad))}" - ) - return self.client.existing_hashes(self.table, sorted(kinds)) - - def sync(self, records: list[VectorRecord], embedder, kinds: set[str]) -> SyncStats: - stats = SyncStats() - existing = self.existing_hashes(kinds) - to_embed: list[VectorRecord] = [] - for record in records: - h = content_hash(record.content) - if record.id not in existing: - to_embed.append(record) - stats.added += 1 - elif existing[record.id] != h: - to_embed.append(record) - stats.updated += 1 - else: - stats.unchanged += 1 - - stats.deleted = 0 - vectors = embedder.embed_documents([r.content for r in to_embed]) if to_embed else [] - rows = [ - { - "record_key": record.id, - "kind": record.kind, - "content_hash": content_hash(record.content), - "metadata": pack_metadata(record), - "embedding": vector, - } - for record, vector in zip(to_embed, vectors) - ] - if rows: - self.client.upsert_records(self.table, rows) - # Gli stale non vengono cancellati in REST writer: restano responsabilita' server-side. - return stats