feat(harness): port vectorstore dual-key + reader RPC (D11, §5.4)

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

@@ -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)

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 '<testo>'}")
+        try:
+            _, fm, body = text.split("---\n", 2)
+        except ValueError as e:
+            raise EvidenceError(f"frontmatter malformato in {path or '<testo>'}") from e
+        meta = yaml.safe_load(fm)
+        if not isinstance(meta, dict):
+            raise EvidenceError(f"frontmatter non valido in {path or '<testo>'}")
+        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 '<testo>'}:\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

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]

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)

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

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.<table_name>`: 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)

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

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,
+    }

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';

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';

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