diff --git a/.superpowers/sdd/evidence-task-5-report.md b/.superpowers/sdd/evidence-task-5-report.md index ad45c3de..e778b59d 100644 --- a/.superpowers/sdd/evidence-task-5-report.md +++ b/.superpowers/sdd/evidence-task-5-report.md @@ -59,3 +59,24 @@ Live pgvector integration needs Docker or an authorized local pgvector endpoint. strategy is logical isolation rather than destructive cleanup because the shared `VectorStore` port intentionally exposes no delete/transaction API; unreachable failed generations can be garbage-collected by a future maintenance job. + +## Review integration wave + +Added an enforceable `metadata_filter` vector-port contract and capability flags. Direct pgvector +places exact Evidence generation/document predicates in SQL before `LIMIT`; HTTP sends the same +filter to the RPC and deliberately does not use the legacy 404 fallback. The reader RPC script now +validates and applies that filter. Normal Evidence search and search-pack use an ACTIVE-aware +searcher that groups active documents by generation, executes complete server-filtered searches, +and merges the results. + +Added exact-generation Evidence cleanup to direct and HTTP writers plus the allowlisted writer RPC. +Pipeline failures compensate both staged filesystem state and vector writes; cleanup failures stay +sanitized and ACTIVE filtering remains the exposure boundary. Corpus-present session artifact +resolution now fails closed on corrupt/missing ACTIVE rather than falling through to source files. + +Focused review-wave verification: 45 passed, scoped Ruff clean. A mocked REST regression proves +the exact filter payload and fail-closed legacy 404 behavior. + +Still outstanding from the expanded review request: Task-4 JobRunner stage-by-stage integration, +published-generation retention/garbage collection, same-fd `dirfd` materialized-file reads, and +live local pgvector integration could not be completed in this wave. diff --git a/harness/scripts/create_vector_reader_rpc.sql b/harness/scripts/create_vector_reader_rpc.sql index 913e3b4c..43c4ba31 100644 --- a/harness/scripts/create_vector_reader_rpc.sql +++ b/harness/scripts/create_vector_reader_rpc.sql @@ -28,7 +28,9 @@ $$; create or replace function public.search_similar( table_name text, query_embedding vector, - limit_count integer + limit_count integer, + kinds text[] default null, + metadata_filter jsonb default null ) returns table(id bigint, similarity real, metadata jsonb) language plpgsql @@ -37,15 +39,31 @@ set search_path = public, vectors, extensions as $$ begin perform public._assert_vector_read_table(table_name); + if metadata_filter is not null and ( + table_name <> 'evidence' + or not (metadata_filter ? 'vector_generation') + or not (metadata_filter ? 'document_ids') + or jsonb_object_length(metadata_filter) <> 2 + or jsonb_typeof(metadata_filter->'document_ids') <> 'array' + ) then + raise exception 'invalid Evidence metadata filter'; + end if; return query execute format( 'select t.id, (1 - (t.embedding <=> $1))::real as similarity, t.metadata from vectors.%I t + where ($3 is null or t.kind = any ($3)) + and ($4 is null or ( + t.metadata->>''vector_generation'' = $4->>''vector_generation'' + and t.metadata->>''document_id'' in ( + select jsonb_array_elements_text($4->''document_ids'') + ) + )) order by t.embedding <=> $1 limit $2', table_name - ) using query_embedding, limit_count; + ) using query_embedding, limit_count, kinds, metadata_filter; end; $$; @@ -75,7 +93,7 @@ end; $$; revoke all on function public._assert_vector_read_table(text) from public; -revoke all on function public.search_similar(text, vector, integer) from public; +revoke all on function public.search_similar(text, vector, integer, text[], jsonb) from public; revoke all on function public.list_tables() from public; -- Revoca dai ruoli client generici, poi abilita solo il reader dedicato @@ -83,15 +101,16 @@ revoke all on function public.list_tables() from public; do $$ begin if exists (select 1 from pg_roles where rolname = 'anon') then - revoke all on function public.search_similar(text, vector, integer) from anon; + revoke all on function public.search_similar(text, vector, integer, text[], jsonb) from anon; revoke all on function public.list_tables() from anon; end if; if exists (select 1 from pg_roles where rolname = 'authenticated') then - revoke all on function public.search_similar(text, vector, integer) from authenticated; + revoke all on function public.search_similar(text, vector, integer, text[], jsonb) from authenticated; revoke all on function public.list_tables() from authenticated; end if; if exists (select 1 from pg_roles where rolname = 'vector_reader') then - grant execute on function public.search_similar(text, vector, integer) to vector_reader; + grant execute on function public.search_similar(text, vector, integer, text[], jsonb) + to vector_reader; grant execute on function public.list_tables() to vector_reader; end if; end $$; diff --git a/harness/scripts/create_vector_writer_rpc.sql b/harness/scripts/create_vector_writer_rpc.sql index 51a1aea5..1fe7d11a 100644 --- a/harness/scripts/create_vector_writer_rpc.sql +++ b/harness/scripts/create_vector_writer_rpc.sql @@ -101,9 +101,30 @@ begin end; $$; +create or replace function public.delete_vector_generation( + table_name text, kind text, generation text +) +returns jsonb +language plpgsql +security definer +set search_path = public, vectors, extensions +as $$ +declare affected integer; +begin + if table_name <> 'evidence' or kind <> 'evidence' or generation !~ '^gen:[0-9a-f]{32}$' then + raise exception 'only an exact Evidence generation may be deleted'; + end if; + delete from vectors.evidence e + where e.kind = 'evidence' and e.metadata->>'vector_generation' = generation; + get diagnostics affected = row_count; + return jsonb_build_object('deleted', affected); +end; +$$; + revoke all on function public._assert_vector_write_table(text, text[]) from public; revoke all on function public.existing_vector_hashes(text, text[]) from public; revoke all on function public.upsert_vector_records(text, jsonb) from public; +revoke all on function public.delete_vector_generation(text, text, text) from public; -- Su alcuni progetti Supabase le funzioni in `public` ricevono grant automatici: revoca -- esplicitamente dai ruoli client generici, poi abilita solo il writer dedicato. @@ -112,14 +133,17 @@ begin if exists (select 1 from pg_roles where rolname = 'anon') then revoke all on function public.existing_vector_hashes(text, text[]) from anon; revoke all on function public.upsert_vector_records(text, jsonb) from anon; + revoke all on function public.delete_vector_generation(text, text, text) from anon; end if; if exists (select 1 from pg_roles where rolname = 'authenticated') then revoke all on function public.existing_vector_hashes(text, text[]) from authenticated; revoke all on function public.upsert_vector_records(text, jsonb) from authenticated; + revoke all on function public.delete_vector_generation(text, text, text) from authenticated; end if; if exists (select 1 from pg_roles where rolname = 'vector_writer') then grant execute on function public.existing_vector_hashes(text, text[]) to vector_writer; grant execute on function public.upsert_vector_records(text, jsonb) to vector_writer; + grant execute on function public.delete_vector_generation(text, text, text) to vector_writer; end if; end $$; diff --git a/harness/tests/test_search_similar_kinds.py b/harness/tests/test_search_similar_kinds.py index 5a16be0a..049d9519 100644 --- a/harness/tests/test_search_similar_kinds.py +++ b/harness/tests/test_search_similar_kinds.py @@ -90,6 +90,25 @@ def test_search_similar_reraises_non_404_with_kinds(monkeypatch): _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 = [] diff --git a/harness/tht/adapters/vector/legacy_direct.py b/harness/tht/adapters/vector/legacy_direct.py index bdec704c..c14202f3 100644 --- a/harness/tht/adapters/vector/legacy_direct.py +++ b/harness/tht/adapters/vector/legacy_direct.py @@ -5,6 +5,7 @@ from sqlalchemy import Engine from tht.ports.vector import ( VectorCapabilities, VectorHealth, + VectorStoreError, VectorWriteRecord, VectorWriteUnavailable, require_positive_limit, @@ -49,8 +50,11 @@ class LegacyDirectVectorStore: *, 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( diff --git a/harness/tht/adapters/vector/pgvector.py b/harness/tht/adapters/vector/pgvector.py index e9980c28..1b1f9970 100644 --- a/harness/tht/adapters/vector/pgvector.py +++ b/harness/tht/adapters/vector/pgvector.py @@ -85,6 +85,8 @@ class PgVectorStore: search=self._reader is not None, existing_hashes=writable, upsert=writable, + metadata_filter=self._reader is not None, + delete_generation=writable, ) def _probe( @@ -242,6 +244,7 @@ class PgVectorStore: *, 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: @@ -262,7 +265,27 @@ class PgVectorStore: ) if kinds and not collection_kinds: continue - where = sql.SQL(" WHERE kind = ANY(%s)") if collection_kinds else sql.SQL("") + 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" + }: + raise VectorStoreError("Unsupported vector metadata filter") + generation = metadata_filter["vector_generation"] + document_ids = metadata_filter["document_ids"] + if not isinstance(generation, str) or not isinstance(document_ids, list): + raise VectorStoreError("Invalid vector metadata filter") + clauses.append(sql.SQL("metadata->>'vector_generation' = %s")) + clauses.append(sql.SQL("metadata->>'document_id' = ANY(%s)")) + filter_params.extend((generation, document_ids)) + 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" @@ -275,8 +298,7 @@ class PgVectorStore: _vector_type(self._schema), ) params = [_vector_literal(embedding)] - if collection_kinds: - params.append(collection_kinds) + 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()) @@ -381,5 +403,30 @@ class PgVectorStore: raw.close() return len(records) + def delete_generation(self, collection: str, generation: str) -> int: + if collection != "evidence" or not generation.startswith("gen:"): + raise VectorStoreError("Only exact Evidence generations may be deleted") + 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" + ).format(_collection(self._schema, collection)), + (generation,), + ) + 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() + __all__ = ["ALLOWED_COLLECTIONS", "PgVectorStore"] diff --git a/harness/tht/adapters/vector/thoth_http.py b/harness/tht/adapters/vector/thoth_http.py index 3efeef85..67379600 100644 --- a/harness/tht/adapters/vector/thoth_http.py +++ b/harness/tht/adapters/vector/thoth_http.py @@ -40,7 +40,8 @@ class ThothHttpVectorStore: def capabilities(self) -> VectorCapabilities: writable = self._writer is not None return VectorCapabilities( - search=self._reader is not None, existing_hashes=writable, upsert=writable + search=self._reader is not None, existing_hashes=writable, upsert=writable, + metadata_filter=self._reader is not None, delete_generation=writable, ) def health(self) -> VectorHealth: @@ -91,6 +92,7 @@ class ThothHttpVectorStore: *, 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: @@ -103,7 +105,13 @@ class ThothHttpVectorStore: for collection in collections: _collection("vectors", collection) try: - rows = self._reader.search_similar(collection, embedding, limit, kinds=kinds) + 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( @@ -144,6 +152,14 @@ class ThothHttpVectorStore: except VectorRestError as exc: raise VectorStoreError(str(exc)) from exc + def delete_generation(self, collection: str, generation: str) -> int: + if collection != "evidence" or not generation.startswith("gen:"): + raise VectorStoreError("Only exact Evidence generations may be deleted") + try: + return self._require_writer().delete_generation(collection, generation) + except VectorRestError as exc: + raise VectorStoreError(str(exc)) from exc + @staticmethod def _row(write_record: VectorWriteRecord) -> dict: record = write_record.record diff --git a/harness/tht/cli/search_cmd.py b/harness/tht/cli/search_cmd.py index 082bc6f6..2766e3cd 100644 --- a/harness/tht/cli/search_cmd.py +++ b/harness/tht/cli/search_cmd.py @@ -48,6 +48,9 @@ def search_cmd( cfg = _load_config_or_exit(config) require_vector_cfg(cfg) + from tht.search.evidence import active_searcher + + runtime_searcher = active_searcher(cfg, open_searcher(cfg)) if kind is not None and kind not in KIND_MAP: typer.secho( f"ERRORE: --kind sconosciuto: {kind} (validi: {', '.join(KIND_MAP)})", @@ -112,7 +115,7 @@ def search_cmd( candidates = combined_search( keyword=keyword, lsh_hits=lsh_hits, - store=open_searcher(cfg), embedder=make_embedder(cfg.embeddings), + store=runtime_searcher, embedder=make_embedder(cfg.embeddings), top=cfg.search.schema_chunk_pool, rrf_k=cfg.search.rrf_k, kinds=KIND_MAP["schema"], ) @@ -156,7 +159,7 @@ def search_cmd( kinds = KIND_MAP.get(kind) if kind else None results = combined_search( keyword=keyword, lsh_hits=lsh_hits if kind != "evidence" else None, - store=open_searcher(cfg), embedder=make_embedder(cfg.embeddings), + store=runtime_searcher, embedder=make_embedder(cfg.embeddings), top=top, rrf_k=cfg.search.rrf_k, kinds=kinds, ) @@ -249,7 +252,9 @@ def pack_cmd( vec = None searcher = embedder = None try: - searcher = open_searcher(cfg) + from tht.search.evidence import active_searcher + + searcher = active_searcher(cfg, open_searcher(cfg)) embedder = make_embedder(cfg.embeddings) vec = embedder.embed_query(question) except degrade as e: diff --git a/harness/tht/cli/vector_cmd.py b/harness/tht/cli/vector_cmd.py index 7a297876..f54546bc 100644 --- a/harness/tht/cli/vector_cmd.py +++ b/harness/tht/cli/vector_cmd.py @@ -60,9 +60,10 @@ def open_searcher(cfg): store = build_vector_store(cfg) class AdapterSearcher: - def search(self, query_vec, top_n=10, kinds=None): + def search(self, query_vec, top_n=10, kinds=None, metadata_filter=None): return store.search( - tables_for_kinds(kinds), query_vec, limit=top_n, kinds=kinds + tables_for_kinds(kinds), query_vec, limit=top_n, kinds=kinds, + metadata_filter=metadata_filter, ) return AdapterSearcher() diff --git a/harness/tht/corpus/pipeline.py b/harness/tht/corpus/pipeline.py index 5da17151..6e0a8ba2 100644 --- a/harness/tht/corpus/pipeline.py +++ b/harness/tht/corpus/pipeline.py @@ -72,6 +72,8 @@ class CorpusPipeline: return self._run(dry_run=dry_run, resume=resume) def _run(self, *, dry_run: bool = False, resume: str | None = None) -> PipelineResult: + generation = None + vector_written = False previous = self.store.active_manifest() try: discovered = self._discover() @@ -138,6 +140,7 @@ class CorpusPipeline: records = [self._vector_record(part, vector, generation) for part, vector in zip(changed_chunks, embeddings, strict=True)] if records: written = self.vector_store.upsert("evidence", records) + vector_written = True if written != len(records): raise PipelineError("vector write count mismatch") generation_path = self.store.generation_path(generation) @@ -156,11 +159,26 @@ class CorpusPipeline: ) self.store.publish(staged) except PipelineError: + self._compensate(generation, vector_written) raise except Exception as error: + self._compensate(generation, vector_written) raise PipelineError("Evidence preprocessing failed") from error return PipelineResult("succeeded", generation, True, changed, unchanged, removed, self.store.manifest(generation)) + def _compensate(self, generation: str | None, vector_written: bool) -> None: + if generation is None: + return + try: + self.store.discard(generation) + except Exception: + pass + if vector_written: + try: + self.vector_store.delete_generation("evidence", generation) + except Exception: + pass + @staticmethod def _vector_record(chunk: CanonicalChunk, embedding: list[float], generation: str): record = VectorRecord( diff --git a/harness/tht/corpus/store.py b/harness/tht/corpus/store.py index 5dd60e0d..1c85c74f 100644 --- a/harness/tht/corpus/store.py +++ b/harness/tht/corpus/store.py @@ -7,6 +7,7 @@ import fcntl import os import re import stat +import shutil import uuid from pathlib import Path from contextlib import contextmanager @@ -137,6 +138,13 @@ class CorpusStore: raise UnsafeCorpusPath("manifest must not be a symlink") return CorpusManifest.model_validate_json(manifest_path.read_text(encoding="utf-8")) + def discard(self, generation: str) -> None: + path = self.generation_path(generation) + if path.exists(): + if path.is_symlink() or not stat.S_ISDIR(path.lstat().st_mode): + raise UnsafeCorpusPath("generation cleanup target is unsafe") + shutil.rmtree(path) + def active_manifest(self) -> CorpusManifest | None: generation = self.active_generation() return self.manifest(generation) if generation else None diff --git a/harness/tht/ports/vector.py b/harness/tht/ports/vector.py index 1b29ea62..84f5b6fe 100644 --- a/harness/tht/ports/vector.py +++ b/harness/tht/ports/vector.py @@ -12,6 +12,8 @@ class VectorCapabilities: search: bool = True existing_hashes: bool = False upsert: bool = False + metadata_filter: bool = False + delete_generation: bool = False @dataclass(frozen=True) @@ -70,12 +72,15 @@ class VectorStore(Protocol): *, limit: int, kinds: list[str] | None = None, + metadata_filter: dict[str, object] | None = None, ) -> list[VectorHit]: ... def existing_hashes(self, collection: str, kinds: list[str]) -> dict[str, str]: ... def upsert(self, collection: str, records: list[VectorWriteRecord]) -> int: ... + def delete_generation(self, collection: str, generation: str) -> int: ... + __all__ = [ "VectorCapabilities", diff --git a/harness/tht/search/evidence.py b/harness/tht/search/evidence.py index e58dca74..a42cb72f 100644 --- a/harness/tht/search/evidence.py +++ b/harness/tht/search/evidence.py @@ -3,6 +3,44 @@ from tht.corpus.store import CorpusStore +class ActiveEvidenceSearcher: + """Searcher facade that enforces ACTIVE generation predicates before LIMIT.""" + + def __init__(self, corpus: CorpusStore, delegate): + self.corpus = corpus + self.delegate = delegate + + def search(self, embedding, top_n=10, kinds=None, metadata_filter=None): + if kinds != ["evidence"]: + return self.delegate.search(embedding, top_n=top_n, kinds=kinds) + manifest = self.corpus.active_manifest() + if manifest is None: + return [] + by_generation: dict[str, list[str]] = {} + mapping = dict(manifest.metadata.get("document_generations", {})) + for document in manifest.documents: + generation = mapping.get(document.document_id, manifest.vector_generation) + if generation: + by_generation.setdefault(generation, []).append(document.document_id) + hits = [] + for generation, document_ids in by_generation.items(): + hits.extend(self.delegate.search( + embedding, top_n=top_n, kinds=["evidence"], + metadata_filter={ + "vector_generation": generation, + "document_ids": document_ids, + }, + )) + return sorted(hits, key=lambda hit: (-hit.similarity, hit.id))[:top_n] + + +def active_searcher(cfg, delegate): + corpus_root = cfg.paths.artifacts.parent / "corpus" + if not corpus_root.exists(): + return delegate + return ActiveEvidenceSearcher(CorpusStore(corpus_root), delegate) + + def active_evidence_hits(store: CorpusStore, vector_store, embedding, *, limit: int): manifest = store.active_manifest() if manifest is None or manifest.vector_generation is None: diff --git a/harness/tht/session/artifacts.py b/harness/tht/session/artifacts.py index 6672ce80..e0e2c44b 100644 --- a/harness/tht/session/artifacts.py +++ b/harness/tht/session/artifacts.py @@ -7,15 +7,11 @@ from tht.session.models import SchemaLinking def _find_evidence_file(evidence_root: Path, evidence_id: str) -> str: # New deployments resolve only immutable materialized files from ACTIVE. Keep # the legacy curated-tree fallback for sessions created before a corpus exists. - try: + corpus_root = evidence_root.parent.parent / "corpus" + if corpus_root.exists(): from tht.corpus.store import CorpusStore from tht.search.evidence import resolve_evidence_file - - corpus_root = evidence_root.parent.parent / "corpus" - if corpus_root.exists(): - return resolve_evidence_file(CorpusStore(corpus_root), evidence_id) - except (OSError, RuntimeError, ValueError): - pass + return resolve_evidence_file(CorpusStore(corpus_root), evidence_id) for match in evidence_root.rglob(f"{evidence_id}.md"): return str(match) return "" diff --git a/harness/tht/vectorstore/rest_client.py b/harness/tht/vectorstore/rest_client.py index 45c006b0..f78db770 100644 --- a/harness/tht/vectorstore/rest_client.py +++ b/harness/tht/vectorstore/rest_client.py @@ -60,6 +60,7 @@ class VectorRestClient: 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` @@ -72,6 +73,13 @@ class VectorRestClient: "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 [] @@ -118,3 +126,12 @@ class VectorRestClient: return int(payload[0]["upserted"]) return len(payload) return len(rows) + + def delete_generation(self, table_name: str, generation: str) -> int: + payload = self._call( + "delete_vector_generation", + {"table_name": table_name, "kind": "evidence", "generation": generation}, + ) + if isinstance(payload, dict): + return int(payload.get("deleted", 0)) + return 0