diff --git a/harness/nsp/db/__init__.py b/harness/nsp/db/__init__.py
new file mode 100644
index 00000000..e69de29b
diff --git a/harness/nsp/db/connection.py b/harness/nsp/db/connection.py
new file mode 100644
index 00000000..e6411ec9
--- /dev/null
+++ b/harness/nsp/db/connection.py
@@ -0,0 +1,37 @@
+from sqlalchemy import Engine, create_engine, text
+
+from nsp.config import DatabaseConfig
+
+
+def make_engine(cfg: DatabaseConfig) -> Engine:
+    url = (
+        f"postgresql+psycopg2://{cfg.user}:{cfg.password}"
+        f"@{cfg.host}:{cfg.port}/{cfg.database}"
+    )
+    return create_engine(url, echo=False)
+
+
+def ping(engine: Engine) -> None:
+    with engine.connect() as conn:
+        conn.execute(text("SELECT 1"))
+
+
+def writable_tables(engine: Engine, schema: str) -> list[str]:
+    """Tabelle dello schema su cui l'utente corrente ha privilegi di scrittura."""
+    q = text("""
+        SELECT c.relname
+        FROM pg_class c
+        JOIN pg_namespace n ON n.oid = c.relnamespace
+        WHERE n.nspname = :schema
+          AND c.relkind IN ('r', 'p')
+          AND has_table_privilege(current_user, c.oid, 'INSERT, UPDATE, DELETE')
+        ORDER BY c.relname
+    """)
+    with engine.connect() as conn:
+        return [row[0] for row in conn.execute(q, {"schema": schema})]
+
+
+def can_create_in_schema(engine: Engine, schema: str) -> bool:
+    q = text("SELECT has_schema_privilege(current_user, :schema, 'CREATE')")
+    with engine.connect() as conn:
+        return bool(conn.execute(q, {"schema": schema}).scalar())
diff --git a/harness/nsp/db/fetch_ca.py b/harness/nsp/db/fetch_ca.py
new file mode 100644
index 00000000..d6b56a89
--- /dev/null
+++ b/harness/nsp/db/fetch_ca.py
@@ -0,0 +1,114 @@
+"""Recupero della catena di certificati presentata da un endpoint HTTPS.
+
+Serve al setup di una postazione *workstation* dietro una CA interna: scarica la
+catena TLS del server REST e la salva in un bundle PEM da puntare con `PSD_SSL_CA`
+(consumato da `requests` via `verify=`). NON installa nulla nel trust store dell'OS.
+"""
+
+from __future__ import annotations
+
+import _ssl
+import socket
+import ssl
+import tempfile
+from pathlib import Path
+from urllib.parse import urlsplit
+
+# Encoding atteso da Certificate.public_bytes() per la catena TLS non verificata.
+_PEM_ENCODING = getattr(_ssl, "ENCODING_PEM", 1)
+
+
+class CaFetchError(Exception):
+    """Errore azionabile durante il recupero della catena CA."""
+
+
+def parse_host_port(base_url: str) -> tuple[str, int]:
+    """Estrae (host, port) da un URL REST https. Porta di default 443."""
+    parts = urlsplit(base_url)
+    if parts.scheme != "https":
+        raise CaFetchError(
+            f"URL non https: {base_url!r}. Il recupero CA ha senso solo su HTTPS."
+        )
+    if not parts.hostname:
+        raise CaFetchError(f"Host mancante nell'URL: {base_url!r}.")
+    return parts.hostname, parts.port or 443
+
+
+def fetch_chain_pem(host: str, port: int = 443, timeout: int = 30) -> list[str]:
+    """Restituisce la catena di certificati presentata da host:port come lista di PEM.
+
+    L'handshake è volutamente *non verificato* (CERT_NONE): stiamo recuperando la catena
+    per poter poi *stabilire* la fiducia, non per fidarci adesso. La verifica vera avviene
+    in seguito quando `PSD_SSL_CA` punta al bundle salvato (es. `nsp db ping`).
+    """
+    ctx = ssl.SSLContext(ssl.PROTOCOL_TLS_CLIENT)
+    ctx.check_hostname = False
+    ctx.verify_mode = ssl.CERT_NONE
+    try:
+        with socket.create_connection((host, port), timeout=timeout) as sock:
+            with ctx.wrap_socket(sock, server_hostname=host) as tls:
+                certs = _unverified_chain(tls)
+    except (OSError, ssl.SSLError) as e:
+        raise CaFetchError(
+            f"Impossibile connettersi a {host}:{port} per recuperare i certificati: {e}"
+        ) from e
+
+    if not certs:
+        raise CaFetchError(
+            f"Nessun certificato presentato da {host}:{port}. "
+            f"In alternativa, estrai la catena a mano con: "
+            f"openssl s_client -showcerts -connect {host}:{port} -servername {host}"
+        )
+    return [_to_pem(c) for c in certs]
+
+
+def _unverified_chain(tls: ssl.SSLSocket) -> list:
+    """Catena presentata dal server. Metodo pubblico su Python >= 3.13, API interna su 3.12."""
+    public = getattr(tls, "get_unverified_chain", None)
+    if public is not None:
+        return list(public() or [])
+    sslobj = getattr(tls, "_sslobj", None)
+    getter = getattr(sslobj, "get_unverified_chain", None) if sslobj is not None else None
+    if getter is None:
+        raise CaFetchError(
+            "Questa versione di Python non espone la catena TLS. "
+            "Estrai la catena a mano con `openssl s_client -showcerts`."
+        )
+    return list(getter() or [])
+
+
+def describe_pem(pem: str) -> str:
+    """Riassunto leggibile (subject / issuer) di un certificato PEM, best-effort.
+
+    Serve a far riconoscere all'utente la CA interna attesa (verifica out-of-band).
+    Restituisce "" se il certificato non è decodificabile.
+    """
+    try:
+        with tempfile.NamedTemporaryFile("w", suffix=".pem", delete=False) as fh:
+            fh.write(pem)
+            tmp = fh.name
+        try:
+            info = _ssl._test_decode_cert(tmp)
+        finally:
+            Path(tmp).unlink(missing_ok=True)
+    except (OSError, ssl.SSLError, ValueError):
+        return ""
+    subject = _name(info.get("subject"))
+    issuer = _name(info.get("issuer"))
+    return f"subject={subject}  issuer={issuer}"
+
+
+def _name(rdns) -> str:
+    """Estrae il CN (o l'intero RDN) da una struttura subject/issuer di _test_decode_cert."""
+    if not rdns:
+        return "?"
+    parts = {k: v for rdn in rdns for (k, v) in rdn}
+    return parts.get("commonName") or ", ".join(f"{k}={v}" for k, v in parts.items())
+
+
+def _to_pem(cert) -> str:
+    """Converte un certificato (_ssl.Certificate o DER bytes) in PEM."""
+    if isinstance(cert, (bytes, bytearray)):
+        return ssl.DER_cert_to_PEM_cert(bytes(cert))
+    pem = cert.public_bytes(_PEM_ENCODING)
+    return pem if isinstance(pem, str) else pem.decode("ascii")
diff --git a/harness/nsp/db/introspect.py b/harness/nsp/db/introspect.py
new file mode 100644
index 00000000..92b6a9da
--- /dev/null
+++ b/harness/nsp/db/introspect.py
@@ -0,0 +1,202 @@
+from datetime import UTC, datetime
+
+from sqlalchemy import Engine, text
+
+from nsp.mschema.models import (
+    ColumnPhysical,
+    ForeignKey,
+    Index,
+    PhysicalSchema,
+    TablePhysical,
+)
+
+# Query adattate da thoth_sqldb2 (Apache 2.0) — vedi src/nsp/vendor/VENDORED.md.
+
+_TABLES_Q = text("""
+    SELECT c.relname AS table_name,
+           COALESCE(d.description, '') AS comment,
+           GREATEST(c.reltuples::bigint, 0) AS row_count
+    FROM pg_class c
+    JOIN pg_namespace n ON n.oid = c.relnamespace
+    LEFT JOIN pg_description d ON d.objoid = c.oid AND d.objsubid = 0
+    WHERE c.relkind IN ('r', 'p') AND n.nspname = :schema
+    ORDER BY c.relname
+""")
+
+_COLUMNS_Q = text("""
+    SELECT a.attname AS column_name,
+           format_type(a.atttypid, a.atttypmod) AS data_type,
+           (NOT a.attnotnull) AS is_nullable,
+           pg_get_expr(d.adbin, d.adrelid) AS column_default,
+           COALESCE(pgd.description, '') AS comment,
+           (ty.typtype = 'e') AS is_enum,
+           EXISTS (
+               SELECT 1 FROM pg_index i
+               WHERE i.indrelid = c.oid AND i.indisprimary AND a.attnum = ANY (i.indkey)
+           ) AS is_pk
+    FROM pg_class c
+    JOIN pg_namespace n ON n.oid = c.relnamespace
+    JOIN pg_attribute a ON a.attrelid = c.oid
+    JOIN pg_type ty ON ty.oid = a.atttypid
+    LEFT JOIN pg_attrdef d ON d.adrelid = c.oid AND d.adnum = a.attnum
+    LEFT JOIN pg_description pgd ON pgd.objoid = c.oid AND pgd.objsubid = a.attnum
+    WHERE c.relname = :table_name AND n.nspname = :schema
+      AND a.attnum > 0 AND NOT a.attisdropped
+    ORDER BY a.attnum
+""")
+
+_FOREIGN_KEYS_Q = text("""
+    SELECT con.conname AS constraint_name,
+           rel.relname AS source_table,
+           a.attname AS source_column,
+           frel.relname AS target_table,
+           fa.attname AS target_column,
+           src.ord
+    FROM pg_constraint con
+    JOIN pg_class rel ON rel.oid = con.conrelid
+    JOIN pg_namespace ns ON ns.oid = rel.relnamespace
+    JOIN pg_class frel ON frel.oid = con.confrelid
+    JOIN unnest(con.conkey) WITH ORDINALITY AS src(attnum, ord) ON true
+    JOIN pg_attribute a ON a.attrelid = con.conrelid AND a.attnum = src.attnum
+    JOIN unnest(con.confkey) WITH ORDINALITY AS dst(attnum, ord) ON dst.ord = src.ord
+    JOIN pg_attribute fa ON fa.attrelid = con.confrelid AND fa.attnum = dst.attnum
+    WHERE con.contype = 'f' AND ns.nspname = :schema
+    ORDER BY rel.relname, con.conname, src.ord
+""")
+
+_INDEXES_Q = text("""
+    SELECT i.relname AS index_name,
+           t.relname AS table_name,
+           ix.indisunique AS is_unique,
+           ix.indisprimary AS is_primary,
+           am.amname AS index_type,
+           array_agg(a.attname ORDER BY a.attnum) AS columns
+    FROM pg_index ix
+    JOIN pg_class i ON i.oid = ix.indexrelid
+    JOIN pg_class t ON t.oid = ix.indrelid
+    JOIN pg_namespace n ON n.oid = t.relnamespace
+    JOIN pg_am am ON am.oid = i.relam
+    JOIN pg_attribute a ON a.attrelid = t.oid AND a.attnum = ANY (ix.indkey)
+    WHERE n.nspname = :schema
+    GROUP BY i.relname, t.relname, ix.indisunique, ix.indisprimary, am.amname
+    ORDER BY t.relname, i.relname
+""")
+
+
+class IntrospectionError(Exception):
+    pass
+
+
+def introspect(engine: Engine, database: str, schema: str) -> PhysicalSchema:
+    with engine.connect() as conn:
+        exists = conn.execute(
+            text("SELECT 1 FROM pg_namespace WHERE nspname = :schema"), {"schema": schema}
+        ).scalar()
+        if not exists:
+            raise IntrospectionError(f"Schema inesistente: {schema}")
+
+        tables: dict[str, TablePhysical] = {}
+        for trow in conn.execute(_TABLES_Q, {"schema": schema}):
+            columns: dict[str, ColumnPhysical] = {}
+            for crow in conn.execute(
+                _COLUMNS_Q, {"table_name": trow.table_name, "schema": schema}
+            ):
+                columns[crow.column_name] = ColumnPhysical(
+                    type=crow.data_type,
+                    nullable=bool(crow.is_nullable),
+                    pk=bool(crow.is_pk),
+                    default=crow.column_default,
+                    comment=crow.comment,
+                    is_enum=bool(crow.is_enum),
+                )
+            tables[trow.table_name] = TablePhysical(
+                comment=trow.comment, row_count=trow.row_count, columns=columns
+            )
+
+        # FK raggruppate per (tabella, constraint), ordinate per posizione
+        grouped: dict[tuple[str, str], ForeignKey] = {}
+        for row in conn.execute(_FOREIGN_KEYS_Q, {"schema": schema}):
+            key = (row.source_table, row.constraint_name)
+            fk = grouped.setdefault(
+                key,
+                ForeignKey(
+                    columns=[], ref_table=row.target_table, ref_columns=[],
+                    name=row.constraint_name,
+                ),
+            )
+            fk.columns.append(row.source_column)
+            fk.ref_columns.append(row.target_column)
+        for (table_name, _), fk in grouped.items():
+            if table_name in tables:
+                tables[table_name].foreign_keys.append(fk)
+
+        for row in conn.execute(_INDEXES_Q, {"schema": schema}):
+            if row.table_name in tables:
+                tables[row.table_name].indexes.append(
+                    Index(
+                        name=row.index_name,
+                        columns=list(row.columns),
+                        unique=bool(row.is_unique),
+                        primary=bool(row.is_primary),
+                        type=row.index_type,
+                    )
+                )
+
+    return PhysicalSchema(
+        database=database,
+        schema=schema,
+        introspected_at=datetime.now(UTC),
+        tables=tables,
+    )
+
+
+def introspect_rest(client, database: str, schema: str) -> PhysicalSchema:
+    """Introspezione via REST (rpc `list_tables`/`table_columns`/`table_comments`/
+    `table_foreign_keys`). Limiti rispetto al diretto: niente indici (nessun rpc) e
+    `is_enum` non disponibile (default False)."""
+    tables: dict[str, TablePhysical] = {}
+    for trow in client.list_tables(schema):
+        if trow.get("type") != "TABLE":
+            continue  # le viste sono fuori scope (come l'introspezione diretta)
+        table_name = trow["table"]
+
+        col_comments = {
+            c["name"]: (c.get("comment") or "")
+            for c in client.table_comments(schema, table_name)
+            if c.get("object") == "COLUMN"
+        }
+        columns: dict[str, ColumnPhysical] = {}
+        for crow in client.table_columns(schema, table_name):
+            name = crow["column"]
+            columns[name] = ColumnPhysical(
+                type=crow["type"],
+                nullable=bool(crow.get("nullable", True)),
+                pk=bool(crow.get("pk", False)),
+                default=crow.get("default"),
+                comment=col_comments.get(name, ""),
+            )
+
+        foreign_keys: list[ForeignKey] = []
+        for fk in client.table_foreign_keys(schema, table_name):
+            foreign_keys.append(
+                ForeignKey(
+                    columns=fk.get("columns") or [fk["column"]],
+                    ref_table=fk.get("ref_table") or fk["target_table"],
+                    ref_columns=fk.get("ref_columns") or [fk["target_column"]],
+                    name=fk.get("name", ""),
+                )
+            )
+
+        tables[table_name] = TablePhysical(
+            comment=trow.get("comment") or "",
+            row_count=int(trow.get("rows") or 0),
+            columns=columns,
+            foreign_keys=foreign_keys,
+        )
+
+    return PhysicalSchema(
+        database=database,
+        schema=schema,
+        introspected_at=datetime.now(UTC),
+        tables=tables,
+    )
diff --git a/harness/nsp/db/sampling.py b/harness/nsp/db/sampling.py
new file mode 100644
index 00000000..0c49717c
--- /dev/null
+++ b/harness/nsp/db/sampling.py
@@ -0,0 +1,160 @@
+import logging
+from dataclasses import dataclass
+
+from sqlalchemy import Engine, text
+
+from nsp.config import ExamplesConfig, LshConfig
+from nsp.mschema.models import Annotations, PhysicalSchema
+
+logger = logging.getLogger(__name__)
+
+TEXT_TYPE_PREFIXES = ("text", "varchar", "character", "char")
+
+
+def is_text_type(pg_type: str) -> bool:
+    return pg_type.lower().startswith(TEXT_TYPE_PREFIXES)
+
+
+def add_examples(engine: Engine, physical: PhysicalSchema, cfg: ExamplesConfig) -> None:
+    """Campiona i valori distinti piu' frequenti delle colonne testuali (in-place)."""
+    schema = physical.db_schema
+    with engine.connect() as conn:
+        for table_name, table in physical.tables.items():
+            for column_name, column in table.columns.items():
+                if not is_text_type(column.type):
+                    continue
+                q = text(f'''
+                    SELECT "{column_name}" FROM (
+                        SELECT "{column_name}", count(*) AS _freq
+                        FROM "{schema}"."{table_name}"
+                        WHERE "{column_name}" IS NOT NULL AND length("{column_name}") > 0
+                        GROUP BY "{column_name}"
+                        ORDER BY _freq DESC
+                        LIMIT :lim
+                    ) AS sub
+                ''')
+                try:
+                    rows = conn.execute(q, {"lim": cfg.max_per_column}).fetchall()
+                except Exception as e:  # colonna non leggibile: si salta, non si interrompe
+                    logger.warning("Campionamento saltato per %s.%s: %s", table_name, column_name, e)
+                    continue
+                column.examples = [str(r[0]) for r in rows]
+
+
+def add_examples_rest(client, physical: PhysicalSchema, cfg: ExamplesConfig) -> None:
+    """Variante REST di add_examples: valori più frequenti via rpc `top_values`."""
+    schema = physical.db_schema
+    for table_name, table in physical.tables.items():
+        for column_name, column in table.columns.items():
+            if not is_text_type(column.type):
+                continue
+            rows = client.top_values(schema, table_name, column_name, cfg.max_per_column)
+            column.examples = [str(r["value"]) for r in rows if r["value"] not in (None, "")]
+
+
+@dataclass
+class SkippedColumn:
+    table: str
+    column: str
+    reason: str
+
+
+@dataclass
+class TruncatedColumn:
+    table: str
+    column: str
+    indexed: int  # quanti valori (i più frequenti) sono stati indicizzati
+
+
+def unique_values_for_lsh(
+    engine: Engine,
+    physical: PhysicalSchema,
+    cfg: LshConfig,
+    annotations: Annotations | None = None,
+) -> tuple[dict[str, dict[str, list[str]]], list[SkippedColumn], list[TruncatedColumn]]:
+    """Valori delle colonne testuali *eligible* per l'indice LSH.
+
+    Indicizza solo colonne con eligibilità effettiva True (le `wide_text` sono escluse:
+    vedi principio di column eligibility). Estrae i valori distinti *più frequenti*
+    (ORDER BY frequenza); se superano `max_values_per_column` la colonna è troncata e
+    segnalata (mai tagliata in silenzio).
+    """
+    from nsp.mschema.eligibility import effective_eligibility
+
+    annotations = annotations or Annotations()
+    schema = physical.db_schema
+    values: dict[str, dict[str, list[str]]] = {}
+    skipped: list[SkippedColumn] = []
+    truncated: list[TruncatedColumn] = []
+    with engine.connect() as conn:
+        for table_name, table in physical.tables.items():
+            table_ann = annotations.tables.get(table_name)
+            for column_name, column in table.columns.items():
+                if not is_text_type(column.type):
+                    continue
+                ann_col = table_ann.columns.get(column_name) if table_ann else None
+                if not effective_eligibility(column, ann_col)[0]:
+                    continue
+                q = text(f'''
+                    SELECT "{column_name}" FROM (
+                        SELECT "{column_name}", count(*) AS _freq
+                        FROM "{schema}"."{table_name}"
+                        WHERE "{column_name}" IS NOT NULL AND length("{column_name}") > 0
+                        GROUP BY "{column_name}"
+                        ORDER BY _freq DESC, "{column_name}"
+                        LIMIT :lim
+                    ) AS sub
+                ''')
+                try:
+                    rows = conn.execute(q, {"lim": cfg.max_values_per_column}).fetchall()
+                except Exception as e:
+                    skipped.append(SkippedColumn(table_name, column_name, f"errore: {e}"))
+                    continue
+                vals = [str(r[0]) for r in rows]
+                if not vals:
+                    continue
+                values.setdefault(table_name, {})[column_name] = vals
+                if len(vals) >= cfg.max_values_per_column:
+                    truncated.append(TruncatedColumn(table_name, column_name, len(vals)))
+    return values, skipped, truncated
+
+
+def unique_values_for_lsh_rest(
+    client,
+    physical: PhysicalSchema,
+    cfg: LshConfig,
+    annotations: Annotations | None = None,
+) -> tuple[dict[str, dict[str, list[str]]], list[SkippedColumn], list[TruncatedColumn]]:
+    """Variante REST di unique_values_for_lsh: valori più frequenti via rpc `top_values`.
+
+    Stessa logica di eligibility e di segnalazione del troncamento del transport diretto.
+    """
+    from nsp.mschema.eligibility import effective_eligibility
+
+    annotations = annotations or Annotations()
+    schema = physical.db_schema
+    values: dict[str, dict[str, list[str]]] = {}
+    skipped: list[SkippedColumn] = []
+    truncated: list[TruncatedColumn] = []
+    for table_name, table in physical.tables.items():
+        table_ann = annotations.tables.get(table_name)
+        for column_name, column in table.columns.items():
+            if not is_text_type(column.type):
+                continue
+            ann_col = table_ann.columns.get(column_name) if table_ann else None
+            if not effective_eligibility(column, ann_col)[0]:
+                continue
+            try:
+                rows = client.top_values(
+                    schema, table_name, column_name, cfg.max_values_per_column
+                )
+            except Exception as e:
+                skipped.append(SkippedColumn(table_name, column_name, f"errore: {e}"))
+                continue
+            vals = [str(r["value"]) for r in rows if r["value"] not in (None, "")]
+            if not vals:
+                continue
+            values.setdefault(table_name, {})[column_name] = vals
+            if len(vals) >= cfg.max_values_per_column:
+                truncated.append(TruncatedColumn(table_name, column_name, len(vals)))
+    return values, skipped, truncated
diff --git a/harness/nsp/mschema/__init__.py b/harness/nsp/mschema/__init__.py
new file mode 100644
index 00000000..e69de29b
diff --git a/harness/nsp/mschema/eligibility.py b/harness/nsp/mschema/eligibility.py
new file mode 100644
index 00000000..e81f2736
--- /dev/null
+++ b/harness/nsp/mschema/eligibility.py
@@ -0,0 +1,93 @@
+"""Classificazione di column eligibility (principio trasversale PsdWp3).
+
+Vedi docs/superpowers/specs/2026-06-13-nsp-column-eligibility-principle.md.
+Il testo ampio (lettere di dimissione, note, anamnesi) è ignorato ovunque; i dati
+provengono solo da numerici, enum, temporali, booleani e testo breve.
+"""
+
+import re
+
+from nsp.config import EligibilityConfig
+from nsp.db.sampling import is_text_type
+from nsp.mschema.models import ColumnAnnotation, ColumnPhysical, PhysicalSchema
+
+_NUMERIC_PREFIXES = (
+    "smallint", "integer", "bigint", "numeric", "decimal", "real", "double", "money",
+)
+_TEMPORAL_PREFIXES = ("date", "time", "timestamp", "interval")
+_LEN_RE = re.compile(r"\((\d+)\)")
+
+
+def _declared_len(pg_type: str) -> int | None:
+    m = _LEN_RE.search(pg_type)
+    return int(m.group(1)) if m else None
+
+
+def classify_column(
+    pg_type: str,
+    is_enum: bool,
+    sampled_avg: float | None,
+    sampled_max: int | None,
+    cfg: EligibilityConfig,
+) -> tuple[bool, str]:
+    """Classifica una colonna come (eligible, reason). Funzione pura, senza I/O."""
+    t = pg_type.strip().lower()
+    if t.endswith("[]"):
+        return False, "wide_text"
+    if is_enum:
+        return True, "enum"
+    if t.startswith(_NUMERIC_PREFIXES):
+        return True, "numeric"
+    if t.startswith("boolean"):
+        return True, "boolean"
+    if t.startswith(_TEMPORAL_PREFIXES):
+        return True, "temporal"
+    if t.startswith("uuid"):
+        return True, "code"
+    if is_text_type(t):
+        declared = _declared_len(t)
+        if declared is not None and declared <= cfg.max_declared_len:
+            return True, "short_text"
+        # bound grande o text/varchar non vincolato: decide il dato campionato
+        if sampled_avg is None or sampled_max is None:
+            return False, "wide_text"
+        if sampled_avg <= cfg.max_avg_length and sampled_max <= cfg.max_sampled_len:
+            return True, "short_text"
+        return False, "wide_text"
+    return False, "wide_text"
+
+
+def _sampled_stats(examples: list[str]) -> tuple[float | None, int | None]:
+    if not examples:
+        return None, None
+    lengths = [len(v) for v in examples]
+    return sum(lengths) / len(lengths), max(lengths)
+
+
+def classify_all(physical: PhysicalSchema, cfg: EligibilityConfig) -> None:
+    """Assegna eligible/eligibility_reason a ogni colonna (in-place) e azzera gli
+    examples delle colonne ignored. Da chiamare DOPO add_examples."""
+    ignore_by_name = {name.lower() for name in cfg.ignore_columns}
+    for table in physical.tables.values():
+        for column_name, column in table.columns.items():
+            if column_name.lower() in ignore_by_name:
+                # colonna di servizio (ETL/audit): ignorata a prescindere dal tipo
+                column.eligible = False
+                column.eligibility_reason = "ignored_by_name"
+                column.examples = []
+                continue
+            avg, mx = _sampled_stats(column.examples)
+            eligible, reason = classify_column(column.type, column.is_enum, avg, mx, cfg)
+            column.eligible = eligible
+            column.eligibility_reason = reason
+            if not eligible:
+                column.examples = []
+
+
+def effective_eligibility(
+    column: ColumnPhysical, annotation: ColumnAnnotation | None
+) -> tuple[bool, str]:
+    """Eligibilità effettiva: l'override in annotations.yaml vince sul fisico."""
+    if annotation is not None and annotation.eligible is not None:
+        return annotation.eligible, "override"
+    return column.eligible, column.eligibility_reason
diff --git a/harness/nsp/mschema/merge.py b/harness/nsp/mschema/merge.py
new file mode 100644
index 00000000..bd7225e7
--- /dev/null
+++ b/harness/nsp/mschema/merge.py
@@ -0,0 +1,15 @@
+from nsp.mschema.models import Annotations, PhysicalSchema
+
+
+def find_orphans(physical: PhysicalSchema, annotations: Annotations) -> list[str]:
+    """Annotazioni che puntano a oggetti spariti dal fisico. Non le rimuove mai."""
+    orphans: list[str] = []
+    for table_name, table_ann in annotations.tables.items():
+        table = physical.tables.get(table_name)
+        if table is None:
+            orphans.append(table_name)
+            continue
+        for column_name in table_ann.columns:
+            if column_name not in table.columns:
+                orphans.append(f"{table_name}.{column_name}")
+    return orphans
diff --git a/harness/nsp/mschema/models.py b/harness/nsp/mschema/models.py
new file mode 100644
index 00000000..256ffb0b
--- /dev/null
+++ b/harness/nsp/mschema/models.py
@@ -0,0 +1,90 @@
+from datetime import datetime
+from pathlib import Path
+from typing import Self
+
+import yaml
+from pydantic import BaseModel, Field
+
+
+class _YamlModel(BaseModel):
+    def to_yaml(self, path: Path) -> None:
+        path.parent.mkdir(parents=True, exist_ok=True)
+        data = self.model_dump(by_alias=True, mode="json", exclude_defaults=False)
+        path.write_text(
+            yaml.safe_dump(data, sort_keys=False, allow_unicode=True, width=120)
+        )
+
+    @classmethod
+    def from_yaml(cls, path: Path) -> Self:
+        raw = yaml.safe_load(path.read_text())
+        return cls.model_validate(raw)
+
+
+class ColumnPhysical(BaseModel):
+    type: str
+    nullable: bool = True
+    pk: bool = False
+    default: str | None = None
+    comment: str = ""
+    examples: list[str] = []
+    is_enum: bool = False
+    eligible: bool = True
+    eligibility_reason: str = ""
+
+
+class ForeignKey(BaseModel):
+    columns: list[str]
+    ref_table: str
+    ref_columns: list[str]
+    name: str = ""
+
+
+class Index(BaseModel):
+    name: str
+    columns: list[str]
+    unique: bool = False
+    primary: bool = False
+    type: str = "btree"
+
+
+class TablePhysical(BaseModel):
+    comment: str = ""
+    row_count: int = 0  # stima da pg_class.reltuples
+    columns: dict[str, ColumnPhysical] = {}
+    foreign_keys: list[ForeignKey] = []
+    indexes: list[Index] = []
+
+
+class PhysicalSchema(_YamlModel):
+    database: str
+    db_schema: str = Field(alias="schema")
+    introspected_at: datetime
+    tables: dict[str, TablePhysical] = {}
+
+    model_config = {"populate_by_name": True}
+
+
+class ColumnAnnotation(BaseModel):
+    description: str = ""
+    synonyms: list[str] = []
+    concepts: list[str] = []
+    evidence: list[str] = []
+    notes: str = ""
+    eligible: bool | None = None
+
+
+class TableAnnotation(BaseModel):
+    description: str = ""
+    concepts: list[str] = []
+    notes: str = ""
+    columns: dict[str, ColumnAnnotation] = {}
+
+
+class Annotations(_YamlModel):
+    tables: dict[str, TableAnnotation] = {}
+
+    @classmethod
+    def from_yaml(cls, path: Path) -> "Annotations":
+        if not path.exists():
+            return cls()
+        return super().from_yaml(path)
diff --git a/harness/nsp/mschema/render.py b/harness/nsp/mschema/render.py
new file mode 100644
index 00000000..95627cbb
--- /dev/null
+++ b/harness/nsp/mschema/render.py
@@ -0,0 +1,141 @@
+from typing import Any
+
+from nsp.mschema.eligibility import effective_eligibility
+from nsp.mschema.models import Annotations, ColumnAnnotation, PhysicalSchema
+
+MAX_EXAMPLES_IN_PROMPT = 5
+
+
+def _ann_col(annotations: Annotations, table: str, column: str) -> ColumnAnnotation | None:
+    ann = annotations.tables.get(table)
+    if ann is None:
+        return None
+    return ann.columns.get(column)
+
+
+def _table_description(physical: PhysicalSchema, annotations: Annotations, table: str) -> str:
+    ann = annotations.tables.get(table)
+    if ann and ann.description:
+        return ann.description
+    return physical.tables[table].comment
+
+
+def _column_description(
+    physical: PhysicalSchema, annotations: Annotations, table: str, column: str
+) -> str:
+    ann = annotations.tables.get(table)
+    if ann and column in ann.columns and ann.columns[column].description:
+        return ann.columns[column].description
+    return physical.tables[table].columns[column].comment
+
+
+def to_mschema_text(
+    physical: PhysicalSchema,
+    annotations: Annotations | None = None,
+    tables: list[str] | None = None,
+) -> str:
+    """Serializzazione testuale in stile ThothAI (【Schema】/【Foreign keys】)."""
+    annotations = annotations or Annotations()
+    selected = [t for t in physical.tables if tables is None or t in tables]
+    lines: list[str] = ["【Schema】"]
+    fk_lines: list[str] = []
+    for table_name in selected:
+        table = physical.tables[table_name]
+        desc = _table_description(physical, annotations, table_name)
+        if desc:
+            lines.append(f"-- {desc}")
+        lines.append(f"CREATE TABLE {table_name} (")
+        for column_name, column in table.columns.items():
+            if not effective_eligibility(column, _ann_col(annotations, table_name, column_name))[0]:
+                continue
+            line = f"    {column_name} {column.type.upper()}"
+            if column.pk:
+                line += " -- PRIMARY KEY"
+            lines.append(line)
+            cdesc = _column_description(physical, annotations, table_name, column_name)
+            if cdesc:
+                lines.append(f"    -- {cdesc}")
+            if column.examples:
+                shown = ", ".join(column.examples[:MAX_EXAMPLES_IN_PROMPT])
+                lines.append(f"    -- Examples: {shown}")
+        lines.append(");")
+        for fk in table.foreign_keys:
+            for src, dst in zip(fk.columns, fk.ref_columns):
+                fk_lines.append(f"{table_name}.{src}={fk.ref_table}.{dst}")
+    lines.extend(["", "【Foreign keys】", *fk_lines])
+    return "\n".join(lines)
+
+
+def to_schema_dict(
+    physical: PhysicalSchema, annotations: Annotations | None = None
+) -> dict[str, Any]:
+    """Vista compatibile con le logiche AV-SQL (schema_dict)."""
+    annotations = annotations or Annotations()
+    out: dict[str, Any] = {}
+    for table_name, table in physical.tables.items():
+        cols = [
+            c
+            for c in table.columns
+            if effective_eligibility(table.columns[c], _ann_col(annotations, table_name, c))[0]
+        ]
+        out[table_name] = {
+            "columns_name": cols,
+            "columns_type": [table.columns[c].type for c in cols],
+            "columns_description": [
+                _column_description(physical, annotations, table_name, c) for c in cols
+            ],
+            "example_values": [table.columns[c].examples for c in cols],
+            "table_to_tablefullname": f"{physical.db_schema}.{table_name}",
+            "primary_keys": [c for c in cols if table.columns[c].pk],
+            "foreign_keys": [
+                {"columns": fk.columns, "ref_table": fk.ref_table, "ref_columns": fk.ref_columns}
+                for fk in table.foreign_keys
+            ],
+        }
+    return out
+
+
+def to_markdown(physical: PhysicalSchema, annotations: Annotations | None = None) -> str:
+    """Report leggibile per il reviewer."""
+    annotations = annotations or Annotations()
+    lines = [
+        f"# Schema {physical.db_schema} ({physical.database})",
+        "",
+        f"Introspezione: {physical.introspected_at.isoformat()} — "
+        f"{len(physical.tables)} tabelle",
+    ]
+    for table_name, table in physical.tables.items():
+        lines += ["", f"## {table_name}", ""]
+        desc = _table_description(physical, annotations, table_name)
+        if desc:
+            lines += [desc, ""]
+        lines += [
+            f"Righe (stima): {table.row_count}",
+            "",
+            "| Colonna | Tipo | Null | PK | Descrizione | Esempi |",
+            "|---|---|---|---|---|---|",
+        ]
+        for column_name, column in table.columns.items():
+            eligible, reason = effective_eligibility(
+                column, _ann_col(annotations, table_name, column_name)
+            )
+            cdesc = _column_description(physical, annotations, table_name, column_name)
+            if not eligible:
+                lines.append(
+                    f"| ~~{column_name}~~ | {column.type} | "
+                    f"{'sì' if column.nullable else 'no'} | "
+                    f"{'sì' if column.pk else ''} | {cdesc} | _ignorata: {reason}_ |"
+                )
+                continue
+            examples = ", ".join(column.examples[:3])
+            lines.append(
+                f"| {column_name} | {column.type} | {'sì' if column.nullable else 'no'} "
+                f"| {'sì' if column.pk else ''} | {cdesc} | {examples} |"
+            )
+        if table.foreign_keys:
+            lines += ["", "Foreign keys:"]
+            for fk in table.foreign_keys:
+                lines.append(
+                    f"- ({', '.join(fk.columns)}) → {fk.ref_table} ({', '.join(fk.ref_columns)})"
+                )
+    return "\n".join(lines)
diff --git a/harness/nsp/rest/__init__.py b/harness/nsp/rest/__init__.py
new file mode 100644
index 00000000..e69de29b
diff --git a/harness/nsp/rest/client.py b/harness/nsp/rest/client.py
new file mode 100644
index 00000000..dde5fa99
--- /dev/null
+++ b/harness/nsp/rest/client.py
@@ -0,0 +1,113 @@
+"""Client per il DWH Chirone esposto via Supabase/PostgREST.
+
+Incapsula i 10 rpc verificati contro produzione. Errori sempre in italiano e azionabili
+(stile `vectorstore/embeddings.py`). Lettura sola: l'API ammette solo SELECT/WITH.
+"""
+
+import requests
+
+from nsp.config import RestConfig
+
+
+class RestError(Exception):
+    """Errore di accesso al DWH via REST, con messaggio leggibile per il reviewer."""
+
+
+class RestClient:
+    def __init__(self, cfg: RestConfig):
+        self.cfg = cfg
+        self._base = cfg.base_url.rstrip("/")
+
+    # -- trasporto -----------------------------------------------------------------
+
+    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 RestError(
+                f"DWH 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"DWH 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 RestError(self._error_msg(fn, resp))
+        if resp.status_code == 204 or not resp.text:
+            return None
+        return resp.json()
+
+    # -- rpc -----------------------------------------------------------------------
+
+    def ping(self) -> dict:
+        return self._call("ping", {})
+
+    def run_query(self, query_text: str) -> list[dict]:
+        """Esegue una SELECT/WITH e restituisce le righe (read-only lato server)."""
+        return self._call("run_query", {"query_text": query_text}) or []
+
+    def explain_query(self, query_text: str) -> list[str]:
+        """EXPLAIN di una SELECT/WITH: ritorna le righe testuali del piano."""
+        rows = self._call("explain_query", {"query_text": query_text}) or []
+        return [r["line"] for r in rows]
+
+    def validate_select(self, query_text: str) -> bool:
+        """True se la query è una SELECT/WITH accettata, False se è una write."""
+        resp = self._post("_validate_select", {"query_text": query_text})
+        if resp.status_code == 204:
+            return True
+        if resp.status_code == 400:
+            return False
+        raise RestError(self._error_msg("_validate_select", resp))
+
+    def list_tables(self, schema_name: str) -> list[dict]:
+        return self._call("list_tables", {"schema_name": schema_name}) or []
+
+    def table_columns(self, schema_name: str, table_name: str) -> list[dict]:
+        return self._call(
+            "table_columns", {"schema_name": schema_name, "table_name": table_name}
+        ) or []
+
+    def table_comments(self, schema_name: str, table_name: str) -> list[dict]:
+        return self._call(
+            "table_comments", {"schema_name": schema_name, "table_name": table_name}
+        ) or []
+
+    def table_foreign_keys(self, schema_name: str, table_name: str) -> list[dict]:
+        return self._call(
+            "table_foreign_keys", {"schema_name": schema_name, "table_name": table_name}
+        ) or []
+
+    def top_values(
+        self, schema_name: str, table_name: str, column_name: str, max_values: int
+    ) -> list[dict]:
+        return self._call(
+            "top_values",
+            {
+                "schema_name": schema_name,
+                "table_name": table_name,
+                "column_name": column_name,
+                "max_values": max_values,
+            },
+        ) or []
+
+    def column_stats(self, schema_name: str, table_name: str, column_name: str) -> dict:
+        return self._call(
+            "column_stats",
+            {"schema_name": schema_name, "table_name": table_name, "column_name": column_name},
+        )
diff --git a/harness/pyproject.toml b/harness/pyproject.toml
index 57358035..11a8ea79 100644
--- a/harness/pyproject.toml
+++ b/harness/pyproject.toml
@@ -32,3 +32,8 @@ line-length = 100
 
 [tool.pytest.ini_options]
 testpaths = ["tests"]
+markers = [
+    "l0: testcontainers tests (need Docker, run locally)",
+    "l2: end-to-end tests requiring real GLM 5.2 + remote DB (skipped when .env incomplete)",
+]
+addopts = "-m 'not l2'"   # L0 runs by default (Docker present); L2 opt-in
diff --git a/harness/tests/conftest.py b/harness/tests/conftest.py
index ef41e039..c055f1c6 100644
--- a/harness/tests/conftest.py
+++ b/harness/tests/conftest.py
@@ -2,6 +2,10 @@ import os
 import sys
 from pathlib import Path
 
+import pytest
+from sqlalchemy import create_engine
+from testcontainers.postgres import PostgresContainer
+
 # Permetti `pytest` lanciato da qualsiasi directory di trovare il package `nsp`
 # (installato in modalità editable nella venv, ma utile anche senza attivazione).
 _ROOT = Path(__file__).resolve().parent.parent
@@ -10,3 +14,31 @@ if str(_ROOT) not in sys.path:
 
 # Directory sessions/ risolta relativamente alla root harness (per i test che creano sessioni)
 os.environ.setdefault("NSP_HARNESS_ROOT", str(_ROOT))
+
+
+# --- L0 fixtures (testcontainers, real Postgres) --------------------------------
+# Session-scoped: un solo container per tutta la run L0. Schema fixture caricato una
+# volta (crea schema dw + dati + ruolo psd_ro read-only). Salta automaticamente se
+# Docker non e' disponibile.
+
+@pytest.fixture(scope="session")
+def pg_container():
+    with PostgresContainer("postgres:16-alpine") as pg:
+        yield pg
+
+
+@pytest.fixture(scope="session")
+def admin_engine(pg_container):
+    engine = create_engine(pg_container.get_connection_url())
+    schema_sql = (Path(__file__).parent / "fixtures" / "schema.sql").read_text()
+    with engine.begin() as conn:
+        conn.exec_driver_sql(schema_sql)
+    yield engine
+    engine.dispose()
+
+
+@pytest.fixture(scope="session")
+def ro_url(pg_container, admin_engine) -> str:
+    host = pg_container.get_container_host_ip()
+    port = pg_container.get_exposed_port(5432)
+    return f"postgresql+psycopg2://psd_ro:psd_ro@{host}:{port}/{pg_container.dbname}"
diff --git a/harness/tests/fixtures/schema.sql b/harness/tests/fixtures/schema.sql
new file mode 100644
index 00000000..8bcaa008
--- /dev/null
+++ b/harness/tests/fixtures/schema.sql
@@ -0,0 +1,58 @@
+CREATE SCHEMA dw;
+
+CREATE TABLE dw.dim_pazienti (
+    id_paziente BIGINT PRIMARY KEY,
+    nome VARCHAR(100),
+    citta VARCHAR(100)
+);
+COMMENT ON TABLE dw.dim_pazienti IS 'Anagrafica pazienti';
+COMMENT ON COLUMN dw.dim_pazienti.citta IS 'Comune di residenza';
+
+CREATE TABLE dw.dim_periodi (
+    anno INT NOT NULL,
+    mese INT NOT NULL,
+    descrizione VARCHAR(50),
+    PRIMARY KEY (anno, mese)
+);
+
+CREATE TYPE dw.tipo_ricovero AS ENUM ('ordinario', 'day_hospital', 'urgenza');
+
+CREATE TABLE dw.fct_ricoveri (
+    id_ricovero BIGINT PRIMARY KEY,
+    id_paziente BIGINT REFERENCES dw.dim_pazienti (id_paziente),
+    anno INT,
+    mese INT,
+    diagnosi VARCHAR(200),
+    procedura TEXT,
+    reparto VARCHAR(50),
+    tipo dw.tipo_ricovero,
+    lettera_dimissione TEXT,
+    etl_last_update TIMESTAMP,
+    FOREIGN KEY (anno, mese) REFERENCES dw.dim_periodi (anno, mese)
+);
+COMMENT ON TABLE dw.fct_ricoveri IS 'Fatti: ricoveri ospedalieri';
+COMMENT ON COLUMN dw.fct_ricoveri.diagnosi IS 'Diagnosi principale (testo)';
+
+INSERT INTO dw.dim_pazienti VALUES
+    (1, 'Mario Rossi', 'Milano'),
+    (2, 'Anna Bianchi', 'Bergamo'),
+    (3, 'Luca Verdi', 'Brescia');
+INSERT INTO dw.dim_periodi VALUES
+    (2025, 1, 'Gennaio 2025'), (2025, 2, 'Febbraio 2025');
+INSERT INTO dw.fct_ricoveri VALUES
+    (10, 1, 2025, 1, 'fibrillazione atriale', 'ablazione transcatetere', 'cardiologia', 'ordinario',
+     'Si dimette il paziente in condizioni cliniche stabili dopo ablazione transcatetere della fibrillazione atriale; si raccomanda terapia anticoagulante orale e controllo cardiologico ambulatoriale a trenta giorni dalla dimissione.',
+     '2025-02-01 03:00:00'),
+    (11, 2, 2025, 1, 'infarto miocardico acuto', 'angioplastica coronarica', 'cardiologia', 'urgenza',
+     'Paziente ricoverato per infarto miocardico acuto trattato con angioplastica coronarica primaria e impianto di stent medicato; decorso post-procedurale regolare, si dimette con doppia antiaggregazione e indicazione a riabilitazione cardiologica.',
+     '2025-02-01 03:00:00'),
+    (12, 3, 2025, 2, 'fibrillazione ventricolare', 'defibrillazione', 'cardiologia', 'urgenza',
+     'Ricovero in urgenza per fibrillazione ventricolare con arresto cardiocircolatorio rianimato; stabilizzato in terapia intensiva cardiologica, si dimette con indicazione a impianto di defibrillatore e follow-up elettrofisiologico.',
+     '2025-03-01 03:00:00'),
+    (13, 1, 2025, 2, 'scompenso cardiaco', NULL, 'pronto soccorso', 'ordinario', NULL, '2025-03-01 03:00:00');
+
+CREATE ROLE psd_ro LOGIN PASSWORD 'psd_ro';
+GRANT USAGE ON SCHEMA dw TO psd_ro;
+GRANT SELECT ON ALL TABLES IN SCHEMA dw TO psd_ro;
+
+ANALYZE;
diff --git a/harness/tests/l0/__init__.py b/harness/tests/l0/__init__.py
new file mode 100644
index 00000000..e69de29b
diff --git a/harness/tests/l0/test_db_connection.py b/harness/tests/l0/test_db_connection.py
new file mode 100644
index 00000000..5871f4b9
--- /dev/null
+++ b/harness/tests/l0/test_db_connection.py
@@ -0,0 +1,56 @@
+"""L0: db/connection read-only enforcement against real Postgres (testcontainers).
+
+The ported read-only contract: the psd_ro role can SELECT but not write, and
+can_create_in_schema / writable_tables reflect that. This is where 'ported code
+is not assumed reliable' gains real teeth for the data layer.
+"""
+import pytest
+from sqlalchemy import create_engine, text
+
+from nsp.db.connection import can_create_in_schema, make_engine, ping, writable_tables
+
+pytestmark = [pytest.mark.l0]
+
+
+def test_ping_succeeds_on_read_only_role(ro_url):
+    engine = create_engine(ro_url)
+    try:
+        ping(engine)  # SELECT 1 — must not raise
+    finally:
+        engine.dispose()
+
+
+def test_read_only_role_cannot_create_in_schema(ro_url):
+    engine = create_engine(ro_url)
+    try:
+        # psd_ro has USAGE + SELECT only, not CREATE on the dw schema.
+        assert can_create_in_schema(engine, "dw") is False
+    finally:
+        engine.dispose()
+
+
+def test_writable_tables_empty_for_read_only_role(ro_url):
+    engine = create_engine(ro_url)
+    try:
+        tables = writable_tables(engine, "dw")
+        assert tables == []  # read-only role has no INSERT/UPDATE/DELETE grants
+    finally:
+        engine.dispose()
+
+
+def test_read_only_role_cannot_insert(ro_url):
+    """The hard guarantee: a write attempt raises (enforced by Postgres, surfaced
+    by our engine)."""
+    engine = create_engine(ro_url)
+    try:
+        with pytest.raises(Exception):
+            with engine.begin() as conn:
+                conn.execute(text('INSERT INTO dw.dim_pazienti VALUES (999, %s, %s)'),
+                             ("test", "test"))
+    finally:
+        engine.dispose()
+
+
+def test_admin_engine_can_create_in_schema(admin_engine):
+    # Sanity: the admin (table owner) CAN create — confirms the test harness itself.
+    assert can_create_in_schema(admin_engine, "dw") is True
diff --git a/harness/tests/l0/test_db_introspect.py b/harness/tests/l0/test_db_introspect.py
new file mode 100644
index 00000000..86b95f16
--- /dev/null
+++ b/harness/tests/l0/test_db_introspect.py
@@ -0,0 +1,58 @@
+"""L0: db/introspect against a known schema (testcontainers).
+Verifies the ported introspection returns the right tables, columns, types, comments,
+FKs, and indexes from a real Postgres catalog.
+"""
+import pytest
+
+from nsp.db.introspect import IntrospectionError, introspect
+
+pytestmark = [pytest.mark.l0]
+
+
+def test_introspect_returns_known_tables(admin_engine):
+    schema = introspect(admin_engine, "testdb", "dw")
+    assert set(schema.tables.keys()) == {"dim_pazienti", "dim_periodi", "fct_ricoveri"}
+    assert schema.db_schema == "dw"
+
+
+def test_introspect_columns_types_and_comments(admin_engine):
+    schema = introspect(admin_engine, "testdb", "dw")
+    pazienti = schema.tables["dim_pazienti"]
+    assert set(pazienti.columns.keys()) == {"id_paziente", "nome", "citta"}
+    assert pazienti.columns["id_paziente"].pk is True
+    assert pazienti.columns["id_paziente"].nullable is False
+    assert "bigint" in pazienti.columns["id_paziente"].type.lower()
+    assert pazienti.comment == "Anagrafica pazienti"
+    assert pazienti.columns["citta"].comment == "Comune di residenza"
+
+
+def test_introspect_enum_detected(admin_engine):
+    schema = introspect(admin_engine, "testdb", "dw")
+    tipo_col = schema.tables["fct_ricoveri"].columns["tipo"]
+    assert tipo_col.is_enum is True
+
+
+def test_introspect_foreign_keys(admin_engine):
+    schema = introspect(admin_engine, "testdb", "dw")
+    ricoveri = schema.tables["fct_ricoveri"]
+    fk_tables = {fk.ref_table for fk in ricoveri.foreign_keys}
+    assert "dim_pazienti" in fk_tables
+    assert "dim_periodi" in fk_tables
+    # the composite FK to dim_periodi (anno, mese)
+    periodi_fk = [fk for fk in ricoveri.foreign_keys if fk.ref_table == "dim_periodi"]
+    assert periodi_fk
+    assert set(periodi_fk[0].columns) == {"anno", "mese"}
+
+
+def test_introspect_indexes(admin_engine):
+    schema = introspect(admin_engine, "testdb", "dw")
+    # composite PK on dim_periodi (anno, mese)
+    periodi = schema.tables["dim_periodi"]
+    pk_indexes = [i for i in periodi.indexes if i.primary]
+    assert pk_indexes, "dim_periodi should have a primary index"
+    assert set(pk_indexes[0].columns) == {"anno", "mese"}
+
+
+def test_introspect_nonexistent_schema_raises(admin_engine):
+    with pytest.raises(IntrospectionError):
+        introspect(admin_engine, "testdb", "nonexistent_schema")
diff --git a/harness/tests/l0/test_db_sampling.py b/harness/tests/l0/test_db_sampling.py
new file mode 100644
index 00000000..dc44bdec
--- /dev/null
+++ b/harness/tests/l0/test_db_sampling.py
@@ -0,0 +1,54 @@
+"""L0: db/sampling against known data (testcontainers).
+Verifies unique_values_for_lsh returns the expected most-frequent values for text
+columns, and that wide_text / non-text columns are excluded.
+"""
+import pytest
+
+from nsp.config import LshConfig
+from nsp.db.introspect import introspect
+from nsp.db.sampling import is_text_type, unique_values_for_lsh
+
+pytestmark = [pytest.mark.l0]
+
+
+def test_is_text_type():
+    assert is_text_type("text")
+    assert is_text_type("varchar(100)")
+    assert is_text_type("character varying")
+    assert not is_text_type("integer")
+    assert not is_text_type("bigint")
+    assert not is_text_type("timestamp without time zone")
+
+
+def test_unique_values_for_lsh_returns_most_frequent(admin_engine):
+    schema = introspect(admin_engine, "testdb", "dw")
+    # Before classify_all, all text columns are eligible=True by default. Sampling
+    # only touches text types regardless.
+    values, skipped, truncated = unique_values_for_lsh(
+        admin_engine, schema, LshConfig(max_values_per_column=100)
+    )
+    # dim_pazienti.citta: Milano, Bergamo, Brescia (3 distinct, all eligible text)
+    citta = values.get("dim_pazienti", {}).get("citta")
+    assert citta is not None
+    assert set(citta) == {"Milano", "Bergamo", "Brescia"}
+
+
+def test_unique_values_for_lsh_excludes_non_text(admin_engine):
+    schema = introspect(admin_engine, "testdb", "dw")
+    values, _, _ = unique_values_for_lsh(
+        admin_engine, schema, LshConfig(max_values_per_column=100)
+    )
+    # id_paziente is bigint — must never appear in the LSH values.
+    assert "id_paziente" not in values.get("dim_pazienti", {})
+
+
+def test_unique_values_for_lsh_truncation_reported(admin_engine):
+    schema = introspect(admin_engine, "testdb", "dw")
+    # Force a tiny cap so procedura/diagnosi columns (which have >2 distinct values)
+    # are reported as truncated rather than silently cut.
+    _, _, truncated = unique_values_for_lsh(
+        admin_engine, schema, LshConfig(max_values_per_column=1)
+    )
+    truncated_cols = {(t.table, t.column) for t in truncated}
+    # fct_ricoveri has several eligible text columns with distinct values
+    assert any(t[0] == "fct_ricoveri" for t in truncated_cols)
diff --git a/harness/tests/test_mschema_eligibility.py b/harness/tests/test_mschema_eligibility.py
new file mode 100644
index 00000000..4319cf04
--- /dev/null
+++ b/harness/tests/test_mschema_eligibility.py
@@ -0,0 +1,109 @@
+"""L1: mschema/eligibility — the column-eligibility principle on fake columns.
+
+Wide text (lettere di dimissione, note, anamnesi) is excluded everywhere; data
+comes only from numerics, enums, temporals, booleans, and short text. Annotation
+override wins over the physical classification.
+"""
+from datetime import datetime
+
+from nsp.config import EligibilityConfig
+from nsp.mschema.eligibility import classify_all, classify_column, effective_eligibility
+from nsp.mschema.models import (
+    Annotations,
+    ColumnAnnotation,
+    ColumnPhysical,
+    PhysicalSchema,
+    TablePhysical,
+)
+
+CFG = EligibilityConfig()
+
+
+# --- classify_column (pure) ----------------------------------------------------
+
+def test_numeric_eligible():
+    ok, reason = classify_column("integer", False, None, None, CFG)
+    assert ok and reason == "numeric"
+
+
+def test_enum_eligible():
+    ok, reason = classify_column("tipo_ricovero", True, None, None, CFG)
+    assert ok and reason == "enum"
+
+
+def test_temporal_eligible():
+    ok, reason = classify_column("timestamp without time zone", False, None, None, CFG)
+    assert ok and reason == "temporal"
+
+
+def test_boolean_eligible():
+    ok, reason = classify_column("boolean", False, None, None, CFG)
+    assert ok and reason == "boolean"
+
+
+def test_short_declared_text_eligible_without_sampling():
+    # varchar(100) <= max_declared_len(128) → eligible without sampling the data
+    ok, reason = classify_column("varchar(100)", False, None, None, CFG)
+    assert ok and reason == "short_text"
+
+
+def test_unbounded_text_without_sampling_is_wide_text():
+    ok, reason = classify_column("text", False, None, None, CFG)
+    assert not ok and reason == "wide_text"
+
+
+def test_long_text_sampled_within_bounds_is_short_text():
+    ok, reason = classify_column("varchar(500)", False, sampled_avg=10.0, sampled_max=50, cfg=CFG)
+    assert ok and reason == "short_text"
+
+
+def test_long_text_sampled_above_bounds_is_wide_text():
+    ok, reason = classify_column("varchar(500)", False, sampled_avg=100.0, sampled_max=600, cfg=CFG)
+    assert not ok and reason == "wide_text"
+
+
+def test_array_type_is_wide_text():
+    ok, reason = classify_column("text[]", False, None, None, CFG)
+    assert not ok and reason == "wide_text"
+
+
+# --- classify_all (in-place) ---------------------------------------------------
+
+def _schema_with(**columns) -> PhysicalSchema:
+    return PhysicalSchema(
+        database="db", schema="dw", introspected_at=datetime(2025, 1, 1),
+        tables={"t": TablePhysical(columns={k: ColumnPhysical(**v) for k, v in columns.items()})},
+    )
+
+
+def test_classify_all_marks_wide_text_and_clears_examples():
+    schema = _schema_with(
+        note=dict(type="text", examples=["a" * 500, "b" * 400]),
+        cod=dict(type="varchar(10)", examples=["X", "Y"]),
+        etl_last_update=dict(type="timestamp", examples=[]),
+    )
+    classify_all(schema, CFG)
+    cols = schema.tables["t"].columns
+    assert cols["note"].eligible is False
+    assert cols["note"].eligibility_reason == "wide_text"
+    assert cols["note"].examples == []  # examples cleared on ignored columns
+    assert cols["cod"].eligible is True
+    # ignore_columns default includes etl_last_update → ignored_by_name regardless of type
+    assert cols["etl_last_update"].eligible is False
+    assert cols["etl_last_update"].eligibility_reason == "ignored_by_name"
+
+
+# --- effective_eligibility (annotation override wins) --------------------------
+
+def test_annotation_override_forces_eligible():
+    col = ColumnPhysical(type="text", eligible=False, eligibility_reason="wide_text")
+    ann = ColumnAnnotation(eligible=True)
+    eff, reason = effective_eligibility(col, ann)
+    assert eff is True
+    assert reason == "override"
+
+
+def test_no_annotation_falls_back_to_physical():
+    col = ColumnPhysical(type="integer", eligible=True, eligibility_reason="numeric")
+    eff, reason = effective_eligibility(col, None)
+    assert eff is True and reason == "numeric"
diff --git a/harness/tests/test_mschema_render.py b/harness/tests/test_mschema_render.py
new file mode 100644
index 00000000..46b6c194
--- /dev/null
+++ b/harness/tests/test_mschema_render.py
@@ -0,0 +1,100 @@
+"""L1: mschema/render — the 3 serialization formats on a fake PhysicalSchema.
+
+Catches port breaks in the render layer (markdown reviewer report, mschema-text
+ThothAI style, schema-dict for AV-SQL). Pure logic, no I/O.
+"""
+from datetime import datetime
+
+from nsp.mschema.models import (
+    Annotations,
+    ColumnAnnotation,
+    ColumnPhysical,
+    ForeignKey,
+    PhysicalSchema,
+    TablePhysical,
+)
+from nsp.mschema.render import to_markdown, to_mschema_text, to_schema_dict
+
+
+def _fake_schema() -> PhysicalSchema:
+    return PhysicalSchema(
+        database="testdb",
+        schema="dw",
+        introspected_at=datetime(2025, 1, 1, 0, 0, 0),
+        tables={
+            "dim_pazienti": TablePhysical(
+                comment="Anagrafica pazienti",
+                row_count=3,
+                columns={
+                    "id_paziente": ColumnPhysical(type="bigint", nullable=False, pk=True),
+                    "citta": ColumnPhysical(type="varchar(100)", examples=["Milano", "Bergamo"]),
+                    "note": ColumnPhysical(type="text", eligible=False,
+                                           eligibility_reason="wide_text"),
+                },
+                foreign_keys=[
+                    ForeignKey(columns=["fk_col"], ref_table="other", ref_columns=["id"]),
+                ],
+            ),
+        },
+    )
+
+
+def test_to_markdown_includes_table_and_marks_ignored():
+    md = to_markdown(_fake_schema())
+    assert "# Schema dw (testdb)" in md
+    assert "## dim_pazienti" in md
+    assert "Anagrafica pazienti" in md
+    # wide_text column is struck through and labeled
+    assert "~~note~~" in md
+    assert "wide_text" in md
+    # foreign key reported
+    assert "other" in md
+
+
+def test_to_mschema_text_style_and_eligibility_filter():
+    txt = to_mschema_text(_fake_schema())
+    assert "【Schema】" in txt
+    assert "【Foreign keys】" in txt
+    assert "CREATE TABLE dim_pazienti (" in txt
+    # eligible columns appear (column name lowercase, type uppercased)
+    assert "id_paziente BIGINT" in txt
+    assert "citta VARCHAR(100)" in txt
+    # wide_text column is filtered out: "note" must not appear as a CREATE TABLE column line
+    assert "    NOTE TEXT" not in txt
+    assert "note" not in txt.replace("【", "").lower().split()
+    # PK marker
+    assert "PRIMARY KEY" in txt
+
+
+def test_to_schema_dict_avsql_shape():
+    d = to_schema_dict(_fake_schema())
+    assert "dim_pazienti" in d
+    entry = d["dim_pazienti"]
+    assert "columns_name" in entry
+    assert "columns_type" in entry
+    assert "primary_keys" in entry
+    assert "foreign_keys" in entry
+    assert "table_to_tablefullname" in entry
+    assert entry["table_to_tablefullname"] == "dw.dim_pazienti"
+    # wide_text column filtered out of columns_name
+    assert "id_paziente" in entry["columns_name"]
+    assert "note" not in entry["columns_name"]
+    assert entry["primary_keys"] == ["id_paziente"]
+
+
+def test_annotations_override_description_in_render():
+    schema = _fake_schema()
+    from nsp.mschema.models import TableAnnotation
+    ann = Annotations(tables={
+        "dim_pazienti": TableAnnotation(columns={
+            "citta": ColumnAnnotation(description="Comune di residenza"),
+        }),
+    })
+    md = to_markdown(schema, ann)
+    assert "Comune di residenza" in md
+
+
+def test_render_subset_of_tables():
+    txt = to_mschema_text(_fake_schema(), tables=["dim_pazienti"])
+    # only the requested table appears
+    assert "dim_pazienti" in txt
diff --git a/harness/tests/test_rest_client.py b/harness/tests/test_rest_client.py
new file mode 100644
index 00000000..f2d76634
--- /dev/null
+++ b/harness/tests/test_rest_client.py
@@ -0,0 +1,115 @@
+"""L1: rest/client RPC contract (mocked transport — no network).
+
+Verifies the ported RestClient: each rpc carries the X-API-Key header, the right
+payload, base_url slash handling, and HTTP/network errors surface as RestError.
+Ported (renamed psdwp3->nsp) from ChironeWp3/tests/test_rest_client.py.
+"""
+import pytest
+import requests
+
+from nsp.config import RestConfig
+from nsp.rest.client import RestClient, RestError
+
+
+class FakeResponse:
+    def __init__(self, payload=None, status=200, text=None):
+        self._payload = payload
+        self.status_code = status
+        self.text = text if text is not None else ("" if payload is None else "json")
+
+    @property
+    def ok(self):
+        return self.status_code < 400
+
+    def json(self):
+        if self._payload is None:
+            raise ValueError("no json")
+        return self._payload
+
+
+def _client(timeout=30):
+    return RestClient(RestConfig(base_url="https://h/dwh/", api_key="dwh_k", timeout=timeout))
+
+
+def _capture(monkeypatch, response):
+    calls = []
+
+    def fake_post(url, json=None, headers=None, timeout=None, verify=None):
+        calls.append({"url": url, "json": json, "headers": headers,
+                      "timeout": timeout, "verify": verify})
+        return response
+
+    monkeypatch.setattr("nsp.rest.client.requests.post", fake_post)
+    return calls
+
+
+def test_run_query_payload_and_rows(monkeypatch):
+    calls = _capture(monkeypatch, FakeResponse([{"x": 1}]))
+    rows = _client().run_query("SELECT 1 AS x")
+    assert rows == [{"x": 1}]
+    c = calls[0]
+    assert c["url"] == "https://h/dwh/rpc/run_query"
+    assert c["json"] == {"query_text": "SELECT 1 AS x"}
+    assert c["headers"]["X-API-Key"] == "dwh_k"
+    assert c["timeout"] == 30
+
+
+def test_explain_query_maps_lines(monkeypatch):
+    plan = [
+        {"line": "Aggregate  (cost=3747.24..3747.25 rows=1 width=8)"},
+        {"line": "  ->  Seq Scan on dim_patient  (cost=0.00..3712.97 rows=13706 width=0)"},
+    ]
+    _capture(monkeypatch, FakeResponse(plan))
+    lines = _client().explain_query("SELECT count(*) FROM dim_patient")
+    assert lines == [
+        "Aggregate  (cost=3747.24..3747.25 rows=1 width=8)",
+        "  ->  Seq Scan on dim_patient  (cost=0.00..3712.97 rows=13706 width=0)",
+    ]
+
+
+def test_ping(monkeypatch):
+    calls = _capture(monkeypatch, FakeResponse({"db_connected": True, "role": "postgres"}))
+    out = _client().ping()
+    assert out["db_connected"] is True
+    assert calls[0]["url"] == "https://h/dwh/rpc/ping"
+    assert calls[0]["json"] == {}
+
+
+def test_top_values_payload(monkeypatch):
+    calls = _capture(monkeypatch, FakeResponse([{"value": "MI", "count": 13737}]))
+    out = _client().top_values("datawarehouse", "dim_patient", "provincia", 1000)
+    assert out == [{"value": "MI", "count": 13737}]
+    assert calls[0]["json"] == {
+        "schema_name": "datawarehouse",
+        "table_name": "dim_patient",
+        "column_name": "provincia",
+        "max_values": 1000,
+    }
+
+
+def test_validate_select_ok_and_write(monkeypatch):
+    _capture(monkeypatch, FakeResponse(status=204))
+    assert _client().validate_select("SELECT 1") is True
+    _capture(monkeypatch, FakeResponse({"message": "Only SELECT / WITH statements are allowed"}, status=400))
+    assert _client().validate_select("DELETE FROM t") is False
+
+
+def test_http_error_surfaces_message(monkeypatch):
+    _capture(monkeypatch, FakeResponse({"message": "boom"}, status=400))
+    with pytest.raises(RestError, match="boom"):
+        _client().run_query("SELECT bad")
+
+
+def test_network_error_actionable(monkeypatch):
+    def boom(*a, **k):
+        raise requests.ConnectionError("refused")
+
+    monkeypatch.setattr("nsp.rest.client.requests.post", boom)
+    with pytest.raises(RestError, match="raggiungibile"):
+        _client().ping()
+
+
+def test_base_url_without_trailing_slash(monkeypatch):
+    calls = _capture(monkeypatch, FakeResponse({"db_connected": True}))
+    RestClient(RestConfig(base_url="https://h/dwh", api_key="k")).ping()
+    assert calls[0]["url"] == "https://h/dwh/rpc/ping"