From 796a39d8935f0736066b18e5ed0c758118938a9f Mon Sep 17 00:00:00 2001 From: mptyl Date: Fri, 26 Jun 2026 22:55:40 +0200 Subject: [PATCH] =?UTF-8?q?feat(harness):=20port=20vectorstore=20dual-key?= =?UTF-8?q?=20+=20reader=20RPC=20(D11,=20=C2=A75.4)?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Ports vectorstore/{rest_client,rest_writer,store,reader,embeddings,records}, evidence/model (leaf dep of records), and cli/_guards (require_vector_write_allowed workstation write-guard). Renamed psdwp3->nsp, verbatim. VectorRestClient gains an api_key property so reader/writer clients carry their distinct keys visibly (spec D11: vector_reader / vector_writer on the same endpoint). scripts/create_vector_reader_rpc.sql is NEW: the reader RPCs (search_similar, list_tables) lived server-side in Supabase and were never versioned. Authored now mirroring the writer allowlist pattern (table allowlist, security definer, revoke from anon/authenticated, grant to vector_reader only). Writer RPC ported verbatim. L1: test_vector_dual_key (7 tests) pins the dual-key construction + the workstation write-guard (exit 4 without writer key). --- harness/nsp/cli/_guards.py | 40 ++++ harness/nsp/evidence/__init__.py | 0 harness/nsp/evidence/model.py | 62 ++++++ harness/nsp/vectorstore/__init__.py | 0 harness/nsp/vectorstore/embeddings.py | 50 +++++ harness/nsp/vectorstore/reader.py | 67 +++++++ harness/nsp/vectorstore/records.py | 101 ++++++++++ harness/nsp/vectorstore/rest_client.py | 104 ++++++++++ harness/nsp/vectorstore/rest_writer.py | 90 +++++++++ harness/nsp/vectorstore/store.py | 191 +++++++++++++++++++ harness/scripts/create_vector_reader_rpc.sql | 99 ++++++++++ harness/scripts/create_vector_writer_rpc.sql | 126 ++++++++++++ harness/tests/test_vector_dual_key.py | 64 +++++++ 13 files changed, 994 insertions(+) create mode 100644 harness/nsp/cli/_guards.py create mode 100644 harness/nsp/evidence/__init__.py create mode 100644 harness/nsp/evidence/model.py create mode 100644 harness/nsp/vectorstore/__init__.py create mode 100644 harness/nsp/vectorstore/embeddings.py create mode 100644 harness/nsp/vectorstore/reader.py create mode 100644 harness/nsp/vectorstore/records.py create mode 100644 harness/nsp/vectorstore/rest_client.py create mode 100644 harness/nsp/vectorstore/rest_writer.py create mode 100644 harness/nsp/vectorstore/store.py create mode 100644 harness/scripts/create_vector_reader_rpc.sql create mode 100644 harness/scripts/create_vector_writer_rpc.sql create mode 100644 harness/tests/test_vector_dual_key.py diff --git a/harness/nsp/cli/_guards.py b/harness/nsp/cli/_guards.py new file mode 100644 index 00000000..05687b54 --- /dev/null +++ b/harness/nsp/cli/_guards.py @@ -0,0 +1,40 @@ +import typer + + +def require_server_profile(cfg, command: str) -> None: + """Rifiuta i comandi di scrittura vectordb sul profilo workstation (exit 4). + + Va chiamata subito dopo il caricamento della config e PRIMA di aprire qualunque + connessione, cosi' su workstation non si tenta mai la connessione diretta al vectordb. + """ + if cfg.profile == "workstation": + typer.secho( + f"ERRORE: `{command}` e' un comando solo-server (scrive nel vectordb centrale). " + f"Sulla postazione locale (profile: workstation) il vectordb si LEGGE via REST, " + f"non si ricostruisce. Esegui questo comando sul server di produzione " + f"(profile: server).", + fg=typer.colors.RED, err=True, + ) + raise typer.Exit(code=4) + + + +def has_vector_write_rest(cfg) -> bool: + return cfg.vector_write_rest is not None and bool(cfg.vector_write_rest.api_key.strip()) + + +def require_vector_write_allowed(cfg, command: str) -> None: + """Permette scritture vectordb da workstation solo con API key REST writer. + + Senza `vector_write_rest`, la workstation resta read-only e i comandi di indexing sono + eseguibili solo sul server con connessione diretta al vectordb. + """ + if cfg.profile == "workstation" and not has_vector_write_rest(cfg): + typer.secho( + f"ERRORE: `{command}` e' un comando solo-server se manca `vector_write_rest`: " + f"scrive nel vectordb centrale. Sulla postazione locale serve la sezione " + f"`vector_write_rest` con una API key di upsert; in alternativa esegui il " + f"comando sul server di produzione (profile: server).", + fg=typer.colors.RED, err=True, + ) + raise typer.Exit(code=4) diff --git a/harness/nsp/evidence/__init__.py b/harness/nsp/evidence/__init__.py new file mode 100644 index 00000000..e69de29b diff --git a/harness/nsp/evidence/model.py b/harness/nsp/evidence/model.py new file mode 100644 index 00000000..c21180eb --- /dev/null +++ b/harness/nsp/evidence/model.py @@ -0,0 +1,62 @@ +from pathlib import Path +from typing import Literal + +import yaml +from pydantic import BaseModel, ValidationError + + +class EvidenceError(Exception): + pass + + +class EvidenceDoc(BaseModel): + id: str + title: str + # tier e status sono opzionali: la sola presenza di un documento basta a + # vettorizzarlo, quindi l'autore ETL non e' obbligato a compilarli. + tier: Literal["structural", "concept"] = "structural" + status: Literal["auto", "draft", "reviewed"] = "reviewed" + sources: list[str] = [] + tables: list[str] = [] + concepts: list[str] = [] + body: str = "" + path: Path | None = None # valorizzato al load, escluso dal dump + + @classmethod + def parse(cls, text: str, path: Path | None = None) -> "EvidenceDoc": + if not text.startswith("---\n"): + raise EvidenceError(f"frontmatter mancante in {path or ''}") + try: + _, fm, body = text.split("---\n", 2) + except ValueError as e: + raise EvidenceError(f"frontmatter malformato in {path or ''}") from e + meta = yaml.safe_load(fm) + if not isinstance(meta, dict): + raise EvidenceError(f"frontmatter non valido in {path or ''}") + try: + return cls.model_validate({**meta, "body": body.strip("\n"), "path": path}) + except ValidationError as e: + raise EvidenceError(f"evidence non valida in {path or ''}:\n{e}") from e + + def dump(self) -> str: + meta = self.model_dump(exclude={"body", "path"}, mode="json") + fm = yaml.safe_dump(meta, sort_keys=False, allow_unicode=True) + return f"---\n{fm}---\n{self.body}\n" + + def save(self, path: Path) -> None: + path.parent.mkdir(parents=True, exist_ok=True) + path.write_text(self.dump()) + self.path = path + + +def load_evidence_dir(root: Path) -> list[EvidenceDoc]: + """Carica ricorsivamente tutte le evidence sotto `root`, preservando la + gerarchia per dominio. I file README (di sola navigazione) sono ignorati.""" + docs: list[EvidenceDoc] = [] + if not root.is_dir(): + return docs + for f in sorted(root.rglob("*.md")): + if f.name.upper().startswith("README"): + continue + docs.append(EvidenceDoc.parse(f.read_text(), path=f)) + return docs diff --git a/harness/nsp/vectorstore/__init__.py b/harness/nsp/vectorstore/__init__.py new file mode 100644 index 00000000..e69de29b diff --git a/harness/nsp/vectorstore/embeddings.py b/harness/nsp/vectorstore/embeddings.py new file mode 100644 index 00000000..131416ff --- /dev/null +++ b/harness/nsp/vectorstore/embeddings.py @@ -0,0 +1,50 @@ +import requests + +from nsp.config import EmbeddingsConfig + +DOC_PREFIX = "search_document: " +QUERY_PREFIX = "search_query: " + + +class EmbeddingsError(Exception): + pass + + +class OllamaEmbeddings: + """Client embeddings via Ollama. Applica i prefissi di task richiesti da nomic v2: + ometterli degrada il retrieval in modo silenzioso.""" + + def __init__(self, cfg: EmbeddingsConfig): + self.cfg = cfg + + def _embed(self, texts: list[str]) -> list[list[float]]: + url = f"{self.cfg.base_url.rstrip('/')}/api/embed" + out: list[list[float]] = [] + for i in range(0, len(texts), self.cfg.batch_size): + batch = texts[i : i + self.cfg.batch_size] + try: + resp = requests.post( + url, json={"model": self.cfg.model, "input": batch}, + timeout=self.cfg.timeout, + ) + resp.raise_for_status() + except requests.RequestException as e: + raise EmbeddingsError( + f"Ollama non raggiungibile su {self.cfg.base_url} " + f"(modello {self.cfg.model}): {e}" + ) from e + embeddings = resp.json().get("embeddings", []) + for v in embeddings: + if len(v) != self.cfg.dim: + raise EmbeddingsError( + f"dimensione embedding inattesa: {len(v)} != {self.cfg.dim} " + f"(modello {self.cfg.model})" + ) + out.extend(embeddings) + return out + + def embed_documents(self, texts: list[str]) -> list[list[float]]: + return self._embed([DOC_PREFIX + t for t in texts]) + + def embed_query(self, text: str) -> list[float]: + return self._embed([QUERY_PREFIX + text])[0] diff --git a/harness/nsp/vectorstore/reader.py b/harness/nsp/vectorstore/reader.py new file mode 100644 index 00000000..f65e0680 --- /dev/null +++ b/harness/nsp/vectorstore/reader.py @@ -0,0 +1,67 @@ +"""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 nsp.vectorstore.rest_client import VectorRestClient +from nsp.vectorstore.store import VectorHit, VectorStore, hit_from_metadata + +# kind PsdWp3 → tabella dello schema `vectors`. +KIND_TO_TABLE = { + "schema_table": "schema_records", + "schema_column": "schema_records", + "evidence": "evidence", + "memory": "memory", +} +ALL_TABLES = ["schema_records", "evidence", "memory"] + + +def tables_for_kinds(kinds: list[str] | None) -> list[str]: + """Tabelle da interrogare per i kind richiesti (tutte se kinds è vuoto/None).""" + if not kinds: + return list(ALL_TABLES) + return sorted({KIND_TO_TABLE[k] for k in kinds if k in KIND_TO_TABLE}) + + +def _merge(hits: list[VectorHit], top_n: int) -> list[VectorHit]: + return sorted(hits, key=lambda h: h.similarity, reverse=True)[:top_n] + + +class RestSearcher: + """Similarity search via REST: una chiamata `search_similar` per tabella, poi fusione.""" + + def __init__(self, client: VectorRestClient): + self.client = client + + def search( + self, query_vec: list[float], top_n: int = 10, kinds: list[str] | None = None + ) -> list[VectorHit]: + hits: list[VectorHit] = [] + for table in tables_for_kinds(kinds): + for row in self.client.search_similar(table, query_vec, top_n): + hits.append(hit_from_metadata(row.get("similarity", 0.0), row.get("metadata"))) + return _merge(hits, top_n) + + +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 + + def search( + self, query_vec: list[float], top_n: int = 10, kinds: list[str] | None = None + ) -> list[VectorHit]: + hits: list[VectorHit] = [] + for table in tables_for_kinds(kinds): + store = VectorStore(self.engine, schema=self.schema, table=table, dim=self.dim) + hits.extend(store.search(query_vec, top_n=top_n)) + return _merge(hits, top_n) diff --git a/harness/nsp/vectorstore/records.py b/harness/nsp/vectorstore/records.py new file mode 100644 index 00000000..91397a77 --- /dev/null +++ b/harness/nsp/vectorstore/records.py @@ -0,0 +1,101 @@ +import re + +from pydantic import BaseModel + +from nsp.evidence.model import EvidenceDoc +from nsp.mschema.models import Annotations, PhysicalSchema + +MAX_EXAMPLES_IN_RECORD = 5 + + +class VectorRecord(BaseModel): + id: str + kind: str # evidence | schema_table | schema_column + ref: str # file/chiave canonica di provenienza + title: str + content: str + metadata: dict = {} + + +def split_markdown(text: str, max_chars: int) -> list[str]: + """Spezza un markdown: intero se sta nel limite, altrimenti per heading '##', + e in ultima istanza per accumulo greedy di righe.""" + if len(text) <= max_chars: + return [text] + parts = re.split(r"(?=^## )", text, flags=re.MULTILINE) + chunks: list[str] = [] + for part in parts: + part = part.strip("\n") + if not part: + continue + if len(part) <= max_chars: + chunks.append(part) + continue + current: list[str] = [] + size = 0 + for line in part.splitlines(): + if size + len(line) > max_chars and current: + chunks.append("\n".join(current)) + current, size = [], 0 + current.append(line) + size += len(line) + 1 + if current: + chunks.append("\n".join(current)) + return chunks + + +def evidence_records(docs: list[EvidenceDoc], max_chunk_chars: int) -> list[VectorRecord]: + """Record per tutte le evidence presenti: la sola presenza basta a indicizzarle.""" + records: list[VectorRecord] = [] + for doc in docs: + content = f"{doc.title}\n\n{doc.body}" + for i, chunk in enumerate(split_markdown(content, max_chunk_chars)): + records.append( + VectorRecord( + id=f"evidence:{doc.id}:{i}", + kind="evidence", + ref=str(doc.path) if doc.path else doc.id, + title=doc.title, + content=chunk, + metadata={ + "status": doc.status, "tier": doc.tier, + "tables": doc.tables, "concepts": doc.concepts, + }, + ) + ) + return records + + +def schema_records(physical: PhysicalSchema, annotations: Annotations) -> list[VectorRecord]: + """Un record per tabella e uno per colonna, da mschema (physical + annotations).""" + records: list[VectorRecord] = [] + for table_name, table in physical.tables.items(): + ann_t = annotations.tables.get(table_name) + t_desc = (ann_t.description if ann_t and ann_t.description else table.comment) + t_concepts = ann_t.concepts if ann_t else [] + lines = [f"Tabella {table_name}", t_desc] + if t_concepts: + lines.append("Concetti: " + ", ".join(t_concepts)) + lines.append("Colonne: " + ", ".join(table.columns)) + records.append( + VectorRecord( + id=f"schema_table:{table_name}", kind="schema_table", ref=table_name, + title=table_name, content="\n".join(filter(None, lines)), + ) + ) + for column_name, column in table.columns.items(): + ann_c = ann_t.columns.get(column_name) if ann_t else None + c_desc = (ann_c.description if ann_c and ann_c.description else column.comment) + lines = [f"Colonna {table_name}.{column_name} ({column.type})", c_desc] + if ann_c and ann_c.synonyms: + lines.append("Sinonimi: " + ", ".join(ann_c.synonyms)) + if column.examples: + lines.append("Esempi: " + ", ".join(column.examples[:MAX_EXAMPLES_IN_RECORD])) + records.append( + VectorRecord( + id=f"schema_column:{table_name}.{column_name}", kind="schema_column", + ref=f"{table_name}.{column_name}", title=f"{table_name}.{column_name}", + content="\n".join(filter(None, lines)), + ) + ) + return records diff --git a/harness/nsp/vectorstore/rest_client.py b/harness/nsp/vectorstore/rest_client.py new file mode 100644 index 00000000..33472a2c --- /dev/null +++ b/harness/nsp/vectorstore/rest_client.py @@ -0,0 +1,104 @@ +"""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 requests + +from nsp.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.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 Exception: + 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 + ) -> list[dict]: + """Ricerca per similarità coseno su `vectors.`: ritorna le righe + `{id, similarity, metadata}` ordinate per similarity decrescente.""" + return self._call( + "search_similar", + { + "query_embedding": query_embedding, + "limit_count": limit_count, + "table_name": table_name, + }, + ) 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))) + return len(payload) if isinstance(payload, list) else len(rows) diff --git a/harness/nsp/vectorstore/rest_writer.py b/harness/nsp/vectorstore/rest_writer.py new file mode 100644 index 00000000..c5474e5b --- /dev/null +++ b/harness/nsp/vectorstore/rest_writer.py @@ -0,0 +1,90 @@ +"""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 nsp.vectorstore.records import VectorRecord +from nsp.vectorstore.rest_client import VectorRestClient +from nsp.vectorstore.store import SyncStats, content_hash + + +KIND_TO_TABLE = { + "schema_table": "schema_records", + "schema_column": "schema_records", + "evidence": "evidence", + "memory": "memory", +} +TABLE_TO_KINDS = { + "schema_records": {"schema_table", "schema_column"}, + "evidence": {"evidence"}, + "memory": {"memory"}, +} + + +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 diff --git a/harness/nsp/vectorstore/store.py b/harness/nsp/vectorstore/store.py new file mode 100644 index 00000000..8f7fe44b --- /dev/null +++ b/harness/nsp/vectorstore/store.py @@ -0,0 +1,191 @@ +import hashlib +import json +from dataclasses import dataclass + +from sqlalchemy import Engine, text + +from nsp.vectorstore.records import VectorRecord + + +def content_hash(content: str) -> str: + return hashlib.sha256(content.encode()).hexdigest() + + +def _to_vector_literal(vec: list[float]) -> str: + return "[" + ",".join(f"{x:.8f}" for x in vec) + "]" + + +@dataclass +class SyncStats: + added: int = 0 + updated: int = 0 + deleted: int = 0 + unchanged: int = 0 + + +@dataclass +class VectorHit: + id: str + kind: str + ref: str + title: str + content: str + metadata: dict + similarity: float + + +def hit_from_metadata(similarity: float, metadata: dict | None) -> VectorHit: + """Ricostruisce un VectorHit dal solo `metadata` (più la similarity). È l'unico modo + disponibile leggendo via REST (`search_similar` ritorna id/similarity/metadata), e viene + usato anche dalla lettura diretta per avere un'unica logica. Tollerante: usa default sui + campi assenti (es. metadata estranei della tabella fake remota).""" + md = metadata or {} + return VectorHit( + id=md.get("record_key", ""), + kind=md.get("kind", ""), + ref=md.get("ref", ""), + title=md.get("title", ""), + content=md.get("content", ""), + metadata=md, + similarity=float(similarity), + ) + + +class VectorStore: + """Tabella pgvector table-scoped: scrittura diretta (loading) su una tabella dello schema + `vectors`. La lettura via REST avviene su `search_similar`; questo store serve al loading e + alla lettura diretta (dev/test). Il contratto della tabella remota richiede `id` (BIGSERIAL), + `embedding vector(N)` e `metadata jsonb`; le colonne extra (`record_key`, `kind`, + `content_hash`) servono solo al loader e non sono esposte dalla REST.""" + + def __init__( + self, engine: Engine, schema: str = "vectors", table: str = "records", dim: int = 768 + ): + self.engine = engine + self.schema = schema + self.dim = dim + self._table = f"{schema}.{table}" + + def init_schema(self) -> None: + with self.engine.begin() as conn: + conn.execute(text("CREATE EXTENSION IF NOT EXISTS vector")) + conn.execute(text(f"CREATE SCHEMA IF NOT EXISTS {self.schema}")) + conn.execute(text(f""" + CREATE TABLE IF NOT EXISTS {self._table} ( + id bigserial PRIMARY KEY, + record_key text UNIQUE NOT NULL, + kind text NOT NULL, + content_hash text NOT NULL, + metadata jsonb NOT NULL DEFAULT '{{}}', + embedding vector({self.dim}) NOT NULL, + indexed_at timestamptz NOT NULL DEFAULT now() + ) + """)) + conn.execute(text( + f"CREATE INDEX IF NOT EXISTS {self._idx('embedding')} ON {self._table} " + f"USING hnsw (embedding vector_cosine_ops) WITH (m = 16, ef_construction = 200)" + )) + conn.execute(text( + f"CREATE INDEX IF NOT EXISTS {self._idx('kind')} ON {self._table} (kind)" + )) + # GRANT al ruolo di sola lettura della REST, solo se esiste (assente in test/locale). + conn.execute(text(f""" + DO $$ BEGIN + IF EXISTS (SELECT 1 FROM pg_roles WHERE rolname = 'vector_reader') THEN + EXECUTE 'GRANT SELECT ON {self._table} TO vector_reader'; + END IF; + END $$; + """)) + + def _idx(self, suffix: str) -> str: + return f"{self._table.replace('.', '_')}_{suffix}_idx" + + def clear(self) -> None: + with self.engine.begin() as conn: + conn.execute(text(f"DELETE FROM {self._table}")) + + def existing_hashes(self, kinds: set[str]) -> dict[str, str]: + q = text( + f"SELECT record_key, content_hash FROM {self._table} WHERE kind = ANY(:kinds)" + ) + with self.engine.connect() as conn: + return dict(conn.execute(q, {"kinds": list(kinds)}).fetchall()) + + def sync(self, records: list[VectorRecord], embedder, kinds: set[str]) -> SyncStats: + """Allinea l'indice ai record correnti (per i kind dati): embedda solo il nuovo + o il modificato, elimina cio' che non esiste piu'.""" + stats = SyncStats() + existing = self.existing_hashes(kinds) + current_ids = {r.id for r in records} + + to_embed: list[VectorRecord] = [] + for r in records: + h = content_hash(r.content) + if r.id not in existing: + to_embed.append(r) + stats.added += 1 + elif existing[r.id] != h: + to_embed.append(r) + stats.updated += 1 + else: + stats.unchanged += 1 + + vectors = embedder.embed_documents([r.content for r in to_embed]) if to_embed else [] + + upsert = text(f""" + INSERT INTO {self._table} + (record_key, kind, content_hash, metadata, embedding) + VALUES + (:record_key, :kind, :content_hash, CAST(:metadata AS jsonb), + CAST(:embedding AS vector)) + ON CONFLICT (record_key) DO UPDATE SET + kind = EXCLUDED.kind, content_hash = EXCLUDED.content_hash, + metadata = EXCLUDED.metadata, embedding = EXCLUDED.embedding, + indexed_at = now() + """) + stale = [i for i in existing if i not in current_ids] + with self.engine.begin() as conn: + for r, vec in zip(to_embed, vectors): + conn.execute(upsert, { + "record_key": r.id, "kind": r.kind, + "content_hash": content_hash(r.content), + "metadata": json.dumps(_pack_metadata(r)), + "embedding": _to_vector_literal(vec), + }) + if stale: + conn.execute( + text(f"DELETE FROM {self._table} WHERE record_key = ANY(:ids)"), + {"ids": stale}, + ) + stats.deleted = len(stale) + return stats + + def search( + self, query_vec: list[float], top_n: int = 10, kinds: list[str] | None = None + ) -> list[VectorHit]: + where = "WHERE kind = ANY(:kinds)" if kinds else "" + q = text(f""" + SELECT metadata, 1 - (embedding <=> CAST(:q AS vector)) AS similarity + FROM {self._table} + {where} + ORDER BY embedding <=> CAST(:q AS vector) + LIMIT :top_n + """) + params: dict = {"q": _to_vector_literal(query_vec), "top_n": top_n} + if kinds: + params["kinds"] = kinds + with self.engine.connect() as conn: + rows = conn.execute(q, params).fetchall() + return [hit_from_metadata(r.similarity, r.metadata) for r in rows] + + +def _pack_metadata(r: VectorRecord) -> dict: + """Impacchetta nel `metadata` (unica colonna letta via REST) tutta la semantica PsdWp3.""" + return { + "kind": r.kind, + "ref": r.ref, + "record_key": r.id, + "title": r.title, + "content": r.content, + **r.metadata, + } diff --git a/harness/scripts/create_vector_reader_rpc.sql b/harness/scripts/create_vector_reader_rpc.sql new file mode 100644 index 00000000..913e3b4c --- /dev/null +++ b/harness/scripts/create_vector_reader_rpc.sql @@ -0,0 +1,99 @@ +-- RPC reader per pgvector via Supabase/PostgREST (spec D11, §5.4). +-- Eseguire sul database Supabase centrale come owner dello schema `vectors`. +-- Prerequisiti: +-- - schema `vectors` con tabelle schema_records/evidence/memory +-- - colonne: record_key text unique, kind text, content_hash text, metadata jsonb, +-- embedding vector(N), indexed_at timestamptz +-- - ruolo/key PostgREST dedicato `vector_reader` (lettura: search_similar, list_tables) +-- +-- A differenza del writer RPC (che era gia' nel repo), i reader RPC vivevano lato server +-- in Supabase e non erano mai stati versionati. Li si riporta qui per completezza del +-- modello dual-key: la key reader riceve solo EXECUTE su queste due funzioni, niente +-- SELECT raw sulle tabelle. Il cleanup/rebuild resta solo-server (vedi writer RPC). + +create or replace function public._assert_vector_read_table(table_name text) +returns void +language plpgsql +stable +as $$ +begin + if table_name not in ('schema_records', 'evidence', 'memory') then + raise sqlstate 'PGRST' using + message = json_build_object('message', 'Unsupported vector table')::text, + detail = json_build_object('status', 400)::text; + end if; +end; +$$; + +create or replace function public.search_similar( + table_name text, + query_embedding vector, + limit_count integer +) +returns table(id bigint, similarity real, metadata jsonb) +language plpgsql +security definer +set search_path = public, vectors, extensions +as $$ +begin + perform public._assert_vector_read_table(table_name); + return query execute format( + 'select t.id, + (1 - (t.embedding <=> $1))::real as similarity, + t.metadata + from vectors.%I t + order by t.embedding <=> $1 + limit $2', + table_name + ) using query_embedding, limit_count; +end; +$$; + +create or replace function public.list_tables() +returns table(table_name text, vector_dimensions integer, record_count bigint) +language plpgsql +security definer +set search_path = public, vectors, extensions +as $$ +begin + return query + select c.relname::text as table_name, + -- dimensione ricavata dalla colonna embedding (vector(N)) se presente + coalesce( + (select (regexp_matches(format_type(a.atttypid, a.atttypmod), 'vector\((\d+)\)'))[1])::int, + 0 + ) as vector_dimensions, + c.reltuples::bigint as record_count + 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' + where n.nspname = 'vectors' + and c.relkind in ('r', 'p') + and c.relname in ('schema_records', 'evidence', 'memory') + order by c.relname; +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.list_tables() from public; + +-- Revoca dai ruoli client generici, poi abilita solo il reader dedicato +-- (e il writer, che usa existing_vector_hashes per l'upsert incrementale). +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.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.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.list_tables() to vector_reader; + end if; +end $$; + +notify pgrst, 'reload schema'; diff --git a/harness/scripts/create_vector_writer_rpc.sql b/harness/scripts/create_vector_writer_rpc.sql new file mode 100644 index 00000000..51a1aea5 --- /dev/null +++ b/harness/scripts/create_vector_writer_rpc.sql @@ -0,0 +1,126 @@ +-- RPC writer controllate per pgvector via Supabase/PostgREST. +-- Eseguire sul database Supabase centrale come owner dello schema `vectors`. +-- Prerequisiti: +-- - schema `vectors` con tabelle schema_records/evidence/memory +-- - colonne: record_key text unique, kind text, content_hash text, metadata jsonb, +-- embedding vector(N), indexed_at timestamptz +-- - ruolo/key PostgREST dedicato `vector_writer` +-- +-- La key writer deve ricevere solo EXECUTE su queste funzioni, non privilegi raw di DELETE +-- sulle tabelle. Cleanup/rebuild restano via terminale sul server. + +create or replace function public._assert_vector_write_table(table_name text, kinds text[]) +returns void +language plpgsql +stable +as $$ +begin + if table_name not in ('schema_records', 'evidence', 'memory') then + raise sqlstate 'PGRST' using + message = json_build_object('message', 'Unsupported vector table')::text, + detail = json_build_object('status', 400)::text; + end if; + + if table_name = 'schema_records' + and exists (select 1 from unnest(kinds) k where k not in ('schema_table', 'schema_column')) then + raise sqlstate 'PGRST' using + message = json_build_object('message', 'Invalid kind for schema_records')::text, + detail = json_build_object('status', 400)::text; + elsif table_name = 'evidence' + and exists (select 1 from unnest(kinds) k where k <> 'evidence') then + raise sqlstate 'PGRST' using + message = json_build_object('message', 'Invalid kind for evidence')::text, + detail = json_build_object('status', 400)::text; + elsif table_name = 'memory' + and exists (select 1 from unnest(kinds) k where k <> 'memory') then + raise sqlstate 'PGRST' using + message = json_build_object('message', 'Invalid kind for memory')::text, + detail = json_build_object('status', 400)::text; + end if; +end; +$$; + +create or replace function public.existing_vector_hashes(table_name text, kinds text[]) +returns table(record_key text, content_hash text) +language plpgsql +security definer +set search_path = public, vectors, extensions +as $$ +begin + perform public._assert_vector_write_table(table_name, kinds); + return query execute format( + 'select record_key, content_hash from vectors.%I where kind = any ($1)', + table_name + ) using kinds; +end; +$$; + +create or replace function public.upsert_vector_records(table_name text, rows jsonb) +returns jsonb +language plpgsql +security definer +set search_path = public, vectors, extensions +as $$ +declare + r jsonb; + affected integer := 0; + kinds text[]; +begin + if jsonb_typeof(rows) <> 'array' then + raise sqlstate 'PGRST' using + message = json_build_object('message', 'rows must be a JSON array')::text, + detail = json_build_object('status', 400)::text; + end if; + + select coalesce(array_agg(distinct value->>'kind'), array[]::text[]) + into kinds + from jsonb_array_elements(rows) value; + perform public._assert_vector_write_table(table_name, kinds); + + for r in select * from jsonb_array_elements(rows) loop + execute format( + 'insert into vectors.%I (record_key, kind, content_hash, metadata, embedding) + values ($1, $2, $3, $4::jsonb, $5::vector) + on conflict (record_key) do update set + kind = excluded.kind, + content_hash = excluded.content_hash, + metadata = excluded.metadata, + embedding = excluded.embedding, + indexed_at = now()', + table_name + ) using + r->>'record_key', + r->>'kind', + r->>'content_hash', + coalesce(r->'metadata', '{}'::jsonb), + ('[' || (select string_agg(value::text, ',') from jsonb_array_elements(r->'embedding')) || ']'); + affected := affected + 1; + end loop; + + return jsonb_build_object('upserted', 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; + +-- Su alcuni progetti Supabase le funzioni in `public` ricevono grant automatici: revoca +-- esplicitamente dai ruoli client generici, poi abilita solo il writer dedicato. +do $$ +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; + 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; + 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; + end if; +end $$; + +notify pgrst, 'reload schema'; diff --git a/harness/tests/test_vector_dual_key.py b/harness/tests/test_vector_dual_key.py new file mode 100644 index 00000000..e79d98cb --- /dev/null +++ b/harness/tests/test_vector_dual_key.py @@ -0,0 +1,64 @@ +"""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 nsp.cli._guards import has_vector_write_rest, require_vector_write_allowed +from nsp.config import Config, DatabaseConfig, RestConfig +from nsp.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