diff --git a/harness/tht/ctetest.py b/harness/tht/ctetest.py
new file mode 100644
index 00000000..0c9e5ee3
--- /dev/null
+++ b/harness/tht/ctetest.py
@@ -0,0 +1,101 @@
+import json
+from datetime import datetime
+from pathlib import Path
+from typing import Literal
+
+import sqlglot
+from pydantic import BaseModel
+from sqlglot import exp
+
+CTE_TESTS_FILE = "cte_tests.json"
+PROBE = "\nSELECT * FROM __psd_probe__"
+
+
+class CteError(Exception):
+    pass
+
+
+class CteTestRecord(BaseModel):
+    name: str
+    ts: datetime
+    sql_hash: str
+    status: Literal["ok", "error"]
+    columns: list[str] = []
+    row_sample: int = 0
+    execution_ms: int = 0
+    warnings: list[str] = []
+    error: str | None = None
+
+
+def has_trailing_select(cte_sql: str) -> bool:
+    """True se il file, parsato da solo, e' gia' una query completa con SELECT
+    finale dopo il blocco WITH (violazione della regola: il file deve contenere
+    solo WITH ... AS (...)).
+
+    I file CTE corretti contengono solo 'WITH ... AS (...)': sqlglot.parse_one()
+    fallisce su di loro perche' manca la SELECT obbligatoria.
+    Un file con SELECT finale parsa invece correttamente come Select con un nodo With.
+    I file con errori di sintassi falliscono parse_one e ritornano False (no hint)."""
+    try:
+        ast = sqlglot.parse_one(cte_sql, read="postgres")
+    except sqlglot.errors.ParseError:
+        return False
+    return ast is not None and ast.find(exp.With) is not None
+
+
+def last_cte_name(cte_sql: str) -> str:
+    """Nome dell'ultimo CTE di un blocco 'WITH a AS (...), b AS (...)' senza SELECT
+    finale. Trucco AV-SQL: si appende una SELECT fittizia per rendere parsabile."""
+    if not cte_sql.lstrip().upper().startswith("WITH"):
+        raise CteError("il file deve contenere un blocco WITH ... AS (...)")
+    try:
+        ast = sqlglot.parse_one(cte_sql + PROBE, read="postgres")
+    except sqlglot.errors.ParseError as e:
+        raise CteError(f"CTE non parsabile: {e}") from e
+    with_clause = ast.find(exp.With)
+    if with_clause is None or not with_clause.expressions:
+        raise CteError("il file deve contenere un blocco WITH ... AS (...)")
+    return with_clause.expressions[-1].alias
+
+
+def build_test_sql(cte_sql: str) -> str:
+    """SQL completo per il test: il blocco WITH + SELECT * dall'ultimo CTE.
+    Il LIMIT lo inietta run_controlled."""
+    name = last_cte_name(cte_sql)
+    return f"{cte_sql.rstrip().rstrip(',')}\nSELECT * FROM {name}"
+
+
+def _iter_json_objects(text: str):
+    """Estrae gli oggetti JSON di primo livello da `text`, qualunque sia il
+    formato: array JSON, JSONL, o misto (array + righe JSONL appese). Salta
+    spazi, virgole e parentesi d'array fra un oggetto e il successivo."""
+    dec = json.JSONDecoder()
+    i, n = 0, len(text)
+    while i < n:
+        if text[i] in " \t\r\n,[]":
+            i += 1
+            continue
+        obj, end = dec.raw_decode(text, i)
+        yield obj
+        i = end
+
+
+def load_cte_tests(session_dir: Path) -> list[CteTestRecord]:
+    path = session_dir / CTE_TESTS_FILE
+    if not path.exists():
+        return []
+    text = path.read_text()
+    try:
+        objs = list(_iter_json_objects(text))
+    except json.JSONDecodeError as e:
+        raise CteError(f"{CTE_TESTS_FILE} corrotto: {e}") from e
+    return [CteTestRecord.model_validate(o) for o in objs]
+
+
+def append_cte_test(session_dir: Path, record: CteTestRecord) -> None:
+    """Append atomico (una riga JSON per esito): scritture concorrenti sulla
+    stessa sessione non si sovrascrivono, a differenza del rewrite dell'array."""
+    path = session_dir / CTE_TESTS_FILE
+    path.parent.mkdir(parents=True, exist_ok=True)
+    with path.open("a", encoding="utf-8") as f:
+        f.write(record.model_dump_json() + "\n")
diff --git a/harness/tht/datamart.py b/harness/tht/datamart.py
new file mode 100644
index 00000000..ced93374
--- /dev/null
+++ b/harness/tht/datamart.py
@@ -0,0 +1,26 @@
+"""Fase 8 — generazione dbt del datamart (hook, non ancora implementato).
+
+Punto d'innesto unico per la generazione del modello dbt da aggiungere al flusso
+ETL di creazione dei datamart. La logica vera verra' sviluppata in seguito: per
+ora qui vive solo la firma della funzione (il "hook") con le due varianti del
+paziente.
+"""
+
+from pathlib import Path
+
+# Le due varianti previste per il paziente nel datamart:
+# - "clear":        nome, cognome, ecc. in chiaro
+# - "pseudonymized": nome, cognome, ecc. mascherati (pseudoanonimizzati)
+DATAMART_MODES = ("clear", "pseudonymized")
+
+
+def generate_dbt_datamart(session_dir: Path, mode: str) -> None:
+    """Genera il modello dbt del datamart per la sessione (HOOK — stub).
+
+    `mode` sceglie la variante del paziente (vedi DATAMART_MODES). Non ancora
+    implementato: solleva NotImplementedError, intercettata dalla CLI che mostra
+    un messaggio pulito senza interrompere il workflow.
+    """
+    raise NotImplementedError(
+        "Generazione dbt del datamart non ancora implementata"
+    )
diff --git a/harness/tht/execute/__init__.py b/harness/tht/execute/__init__.py
new file mode 100644
index 00000000..f3240e58
--- /dev/null
+++ b/harness/tht/execute/__init__.py
@@ -0,0 +1,102 @@
+import json
+import time
+from dataclasses import dataclass
+
+import sqlglot
+from sqlalchemy import Engine, text
+from sqlalchemy.exc import DBAPIError
+
+
+class ExecutionError(Exception):
+    """Errore di esecuzione con messaggio leggibile per il reviewer."""
+
+
+@dataclass
+class ExecResult:
+    columns: list[str]
+    rows: list[tuple]
+    execution_ms: int
+    truncated: bool
+
+
+@dataclass
+class PlanSummary:
+    total_cost: float
+    plan_rows: int
+    node_types: list[str]
+
+
+def _inject_limit(sql: str, limit: int) -> tuple[str, bool]:
+    """Aggiunge LIMIT limit+1 se assente (il +1 serve a rilevare il troncamento).
+    Se la query ha gia' un suo LIMIT, lo si rispetta."""
+    ast = sqlglot.parse_one(sql, read="postgres")
+    # solo le query (SELECT/UNION & co.) accettano un LIMIT: per qualunque altro
+    # statement non si inietta nulla e la transazione READ ONLY lo rifiutera' (rete 2).
+    if not isinstance(ast, sqlglot.exp.Query) or ast.args.get("limit") is not None:
+        return sql, False
+    return ast.limit(limit + 1).sql(dialect="postgres"), True
+
+
+def _translate_error(e: DBAPIError) -> ExecutionError:
+    msg = str(e.orig)
+    if "statement timeout" in msg or "canceling statement" in msg:
+        return ExecutionError("timeout: la query ha superato statement_timeout_ms")
+    if "read-only" in msg:
+        return ExecutionError("rifiutato: statement non eseguibile in transazione read-only")
+    return ExecutionError(f"errore SQL: {msg}")
+
+
+def run_controlled(engine: Engine, sql: str, *, limit: int, timeout_ms: int) -> ExecResult:
+    """Esecuzione nelle 4 reti: utente RO (a monte), transazione READ ONLY,
+    statement_timeout, LIMIT iniettato via AST."""
+    final_sql, injected = _inject_limit(sql, limit)
+    start = time.monotonic()
+    with engine.connect() as conn:
+        trans = conn.begin()
+        try:
+            conn.execute(text("SET TRANSACTION READ ONLY"))
+            conn.execute(text(f"SET LOCAL statement_timeout = {int(timeout_ms)}"))
+            result = conn.execute(text(final_sql))
+            columns = list(result.keys())
+            rows = [tuple(r) for r in result.fetchall()]
+        except DBAPIError as e:
+            raise _translate_error(e) from e
+        finally:
+            trans.rollback()
+    elapsed_ms = int((time.monotonic() - start) * 1000)
+    truncated = injected and len(rows) > limit
+    return ExecResult(
+        columns=columns, rows=rows[:limit], execution_ms=elapsed_ms, truncated=truncated
+    )
+
+
+def explain(engine: Engine, sql: str, *, timeout_ms: int) -> PlanSummary:
+    """EXPLAIN (FORMAT JSON), mai ANALYZE: il piano si stima, non si esegue."""
+    with engine.connect() as conn:
+        trans = conn.begin()
+        try:
+            conn.execute(text("SET TRANSACTION READ ONLY"))
+            conn.execute(text(f"SET LOCAL statement_timeout = {int(timeout_ms)}"))
+            row = conn.execute(text(f"EXPLAIN (FORMAT JSON) {sql}")).fetchone()
+        except DBAPIError as e:
+            raise _translate_error(e) from e
+        finally:
+            trans.rollback()
+    payload = row[0]
+    if isinstance(payload, str):
+        payload = json.loads(payload)
+    plan = payload[0]["Plan"]
+
+    node_types: list[str] = []
+
+    def collect(node: dict) -> None:
+        node_types.append(node.get("Node Type", "?"))
+        for child in node.get("Plans", []):
+            collect(child)
+
+    collect(plan)
+    return PlanSummary(
+        total_cost=float(plan.get("Total Cost", 0.0)),
+        plan_rows=int(plan.get("Plan Rows", 0)),
+        node_types=node_types,
+    )
diff --git a/harness/tht/execute/warnings.py b/harness/tht/execute/warnings.py
new file mode 100644
index 00000000..820c0f37
--- /dev/null
+++ b/harness/tht/execute/warnings.py
@@ -0,0 +1,40 @@
+from sqlglot import exp
+
+from tht.config import ExecutionConfig
+from tht.execute import ExecResult, PlanSummary
+
+
+def runtime_warnings(result: ExecResult, cfg: ExecutionConfig) -> list[str]:
+    warnings: list[str] = []
+    if not result.rows:
+        warnings.append("risultato vuoto: verificare filtri e join")
+    if result.truncated:
+        warnings.append("righe = limit: risultato troncato dalla preview")
+    if result.execution_ms > cfg.warn_execution_ms:
+        warnings.append(
+            f"tempo di esecuzione elevato: {result.execution_ms} ms "
+            f"(soglia {cfg.warn_execution_ms} ms)"
+        )
+    return warnings
+
+
+def plan_warnings(plan: PlanSummary, cfg: ExecutionConfig) -> list[str]:
+    warnings: list[str] = []
+    if plan.plan_rows > cfg.warn_plan_rows:
+        warnings.append(
+            f"righe stimate dal piano: {plan.plan_rows} (soglia {cfg.warn_plan_rows}) — "
+            f"possibile join molto ampio o cardinalita' anomala"
+        )
+    return warnings
+
+
+def static_warnings(ast: exp.Expression) -> list[str]:
+    """Euristica cast sospetti: CAST su colonne dentro WHERE (non-sargable)."""
+    warnings: list[str] = []
+    for where in ast.find_all(exp.Where):
+        for cast in where.find_all(exp.Cast):
+            if cast.find(exp.Column) is not None:
+                warnings.append(
+                    f"cast su colonna nel WHERE (possibile non-sargable): {cast.sql()}"
+                )
+    return warnings
diff --git a/harness/tht/lshindex/__init__.py b/harness/tht/lshindex/__init__.py
new file mode 100644
index 00000000..957539f9
--- /dev/null
+++ b/harness/tht/lshindex/__init__.py
@@ -0,0 +1,77 @@
+import json
+import pickle
+from dataclasses import dataclass
+from pathlib import Path
+from typing import Any
+
+from datasketch import MinHash, MinHashLSH
+
+from tht.config import LshConfig
+from tht.vendor.thoth_lsh import create_lsh_index, create_minhash
+
+
+class LshIndexError(Exception):
+    pass
+
+
+@dataclass
+class LshHit:
+    table: str
+    column: str
+    value: str
+    score: float
+
+
+def build_index(
+    values: dict[str, dict[str, list[str]]], cfg: LshConfig, verbose: bool = False
+) -> tuple[MinHashLSH, dict[str, tuple[MinHash, str, str, str]]]:
+    return create_lsh_index(
+        values, signature_size=cfg.signature_size, n_gram=cfg.n_gram,
+        threshold=cfg.threshold, verbose=verbose,
+    )
+
+
+def _paths(directory: Path, name: str) -> tuple[Path, Path, Path]:
+    return (
+        directory / f"{name}_lsh.pkl",
+        directory / f"{name}_minhashes.pkl",
+        directory / f"{name}_meta.json",
+    )
+
+
+def save_index(lsh, minhashes, cfg: LshConfig, directory: Path, name: str) -> None:
+    directory.mkdir(parents=True, exist_ok=True)
+    lsh_path, mh_path, meta_path = _paths(directory, name)
+    lsh_path.write_bytes(pickle.dumps(lsh))
+    mh_path.write_bytes(pickle.dumps(minhashes))
+    meta_path.write_text(
+        json.dumps(
+            {"signature_size": cfg.signature_size, "n_gram": cfg.n_gram,
+             "threshold": cfg.threshold, "entries": len(minhashes)},
+            indent=2,
+        )
+    )
+
+
+def load_index(directory: Path, name: str) -> tuple[Any, Any, dict]:
+    lsh_path, mh_path, meta_path = _paths(directory, name)
+    if not (lsh_path.exists() and mh_path.exists() and meta_path.exists()):
+        raise LshIndexError(
+            f"Indice LSH non trovato in {directory} (atteso {name}_lsh.pkl). "
+            f"Esegui prima `nsp lsh build`."
+        )
+    lsh = pickle.loads(lsh_path.read_bytes())
+    minhashes = pickle.loads(mh_path.read_bytes())
+    meta = json.loads(meta_path.read_text())
+    return lsh, minhashes, meta
+
+
+def query_index(lsh, minhashes, keyword: str, meta: dict, top_n: int = 10) -> list[LshHit]:
+    """Query con score: i parametri MinHash vengono dal meta dell'indice, non dalla config."""
+    qmh = create_minhash(meta["signature_size"], keyword, meta["n_gram"])
+    scored = [(key, qmh.jaccard(minhashes[key][0])) for key in lsh.query(qmh)]
+    scored.sort(key=lambda kv: kv[1], reverse=True)
+    return [
+        LshHit(table=minhashes[k][1], column=minhashes[k][2], value=minhashes[k][3], score=s)
+        for k, s in scored[:top_n]
+    ]
diff --git a/harness/tht/report.py b/harness/tht/report.py
new file mode 100644
index 00000000..09c4408f
--- /dev/null
+++ b/harness/tht/report.py
@@ -0,0 +1,74 @@
+from tht.execute import ExecResult, PlanSummary
+from tht.sqlcheck import CheckResult
+
+NOTE_HEADING = "## Note del reviewer"
+
+
+def _markdown_table(columns: list[str], rows: list[tuple]) -> str:
+    lines = [
+        "| " + " | ".join(columns) + " |",
+        "|" + "---|" * len(columns),
+    ]
+    for row in rows:
+        lines.append("| " + " | ".join(str(v) for v in row) + " |")
+    return "\n".join(lines)
+
+
+def render_validation_report(
+    *,
+    session_id: str,
+    check: CheckResult,
+    plan: PlanSummary,
+    plan_warnings: list[str],
+    result: ExecResult,
+    runtime_warnings: list[str],
+    static_warnings: list[str],
+    limit: int,
+    reviewer_notes: str = "",
+) -> str:
+    """Sezioni fattuali del validation report. La sezione note e' di Pi/reviewer:
+    se gia' scritta, va passata in reviewer_notes per preservarla."""
+
+    def warn(ws: list[str]) -> str:
+        return "\n".join(f"- warning: {w}" for w in ws) if ws else "- nessun warning"
+
+    parts = [
+        f"# Validation report — {session_id}",
+        "",
+        "> Avvertenza: questo report contiene sample di dati del database.",
+        "",
+        "## Parsing e read-only",
+        "",
+        "- parsing: OK" if check.ok else "- parsing: ERRORI",
+        "- statement read-only per struttura: OK" if check.ok else
+        "\n".join(f"- ERRORE: {e}" for e in check.errors),
+        warn(check.warnings + static_warnings),
+        "",
+        "## Piano di esecuzione",
+        "",
+        f"- costo totale stimato: {plan.total_cost}",
+        f"- righe stimate: {plan.plan_rows}",
+        f"- nodi: {', '.join(plan.node_types)}",
+        warn(plan_warnings),
+        "",
+        "## Preview / aggregato",
+        "",
+        f"- tempo di esecuzione: {result.execution_ms} ms (limit {limit})",
+        "",
+        _markdown_table(result.columns, result.rows),
+        "",
+        warn(runtime_warnings),
+        "",
+        NOTE_HEADING,
+        "",
+        reviewer_notes.strip(),
+        "",
+    ]
+    return "\n".join(parts)
+
+
+def extract_reviewer_notes(report_md: str) -> str:
+    """Estrae il contenuto della sezione note da un report esistente."""
+    if NOTE_HEADING not in report_md:
+        return ""
+    return report_md.split(NOTE_HEADING, 1)[1].strip()
diff --git a/harness/tht/rest/execute.py b/harness/tht/rest/execute.py
new file mode 100644
index 00000000..07626ec5
--- /dev/null
+++ b/harness/tht/rest/execute.py
@@ -0,0 +1,38 @@
+"""Esecuzione controllata e EXPLAIN via REST (rpc `run_query`/`explain_query`).
+
+Produce gli stessi tipi del transport diretto (`ExecResult`, `PlanSummary`) e traduce
+`RestError` in `ExecutionError`, così le fasi 6/7 e la gestione errori della CLI non cambiano.
+
+Differenze rispetto al diretto: read-only e statement_timeout sono garantiti lato server;
+il client mantiene solo l'iniezione del LIMIT (per il rilevamento del troncamento in preview).
+"""
+
+import time
+
+from tht.execute import ExecResult, ExecutionError, PlanSummary, _inject_limit
+from tht.rest.client import RestError
+from tht.rest.explain import parse_text_plan
+
+
+def run_controlled_rest(client, sql: str, *, limit: int) -> ExecResult:
+    final_sql, injected = _inject_limit(sql, limit)
+    start = time.monotonic()
+    try:
+        rows_dicts = client.run_query(final_sql)
+    except RestError as e:
+        raise ExecutionError(str(e)) from e
+    elapsed_ms = int((time.monotonic() - start) * 1000)
+    columns = list(rows_dicts[0].keys()) if rows_dicts else []
+    rows = [tuple(r.get(c) for c in columns) for r in rows_dicts]
+    truncated = injected and len(rows) > limit
+    return ExecResult(
+        columns=columns, rows=rows[:limit], execution_ms=elapsed_ms, truncated=truncated
+    )
+
+
+def explain_rest(client, sql: str) -> PlanSummary:
+    try:
+        lines = client.explain_query(sql)
+    except RestError as e:
+        raise ExecutionError(str(e)) from e
+    return parse_text_plan(lines)
diff --git a/harness/tht/rest/explain.py b/harness/tht/rest/explain.py
new file mode 100644
index 00000000..e698b501
--- /dev/null
+++ b/harness/tht/rest/explain.py
@@ -0,0 +1,35 @@
+"""Parsing del piano EXPLAIN testuale restituito dall'rpc `explain_query`.
+
+Il transport diretto usa `EXPLAIN (FORMAT JSON)`; via REST l'rpc ritorna il piano come
+righe di testo (formato default di EXPLAIN). Questo modulo le riconduce a `PlanSummary`,
+così la logica a valle (`plan_warnings`, display) non cambia.
+"""
+
+import re
+
+from tht.execute import ExecutionError, PlanSummary
+
+# es. "(cost=0.00..3712.97 rows=13706 width=0)"
+_COST_RE = re.compile(r"\(cost=[\d.]+\.\.([\d.]+)\s+rows=(\d+)\s+width=\d+\)")
+
+
+def parse_text_plan(lines: list[str]) -> PlanSummary:
+    node_types: list[str] = []
+    total_cost = 0.0
+    plan_rows = 0
+    root_seen = False
+    for line in lines:
+        m = _COST_RE.search(line)
+        if m is None:
+            # riga di dettaglio (Filter, Index Cond, …): non è un nodo del piano
+            continue
+        label = line.split("(cost=")[0].replace("->", " ").strip()
+        if label:
+            node_types.append(label)
+        if not root_seen:
+            total_cost = float(m.group(1))
+            plan_rows = int(m.group(2))
+            root_seen = True
+    if not root_seen:
+        raise ExecutionError("piano EXPLAIN vuoto o non interpretabile")
+    return PlanSummary(total_cost=total_cost, plan_rows=plan_rows, node_types=node_types)
diff --git a/harness/tht/sqlcheck/__init__.py b/harness/tht/sqlcheck/__init__.py
new file mode 100644
index 00000000..4a646830
--- /dev/null
+++ b/harness/tht/sqlcheck/__init__.py
@@ -0,0 +1,112 @@
+from dataclasses import dataclass, field
+
+import sqlglot
+from sqlglot import exp
+
+from tht.config import ExecutionConfig
+from tht.mschema.models import PhysicalSchema
+
+DIALECT = "postgres"
+
+FORBIDDEN_NODES: dict[type, str] = {
+    exp.Insert: "INSERT",
+    exp.Update: "UPDATE",
+    exp.Delete: "DELETE",
+    exp.Create: "CREATE",
+    exp.Drop: "DROP",
+    exp.Alter: "ALTER",
+    exp.Merge: "MERGE",
+    exp.TruncateTable: "TRUNCATE",
+    exp.Grant: "GRANT",
+    exp.Command: "comando non-SELECT",  # CALL e altri comandi opachi
+}
+
+DEFAULT_FORBIDDEN_FUNCTIONS = set(ExecutionConfig().forbidden_functions)
+
+
+@dataclass
+class CheckResult:
+    ok: bool
+    errors: list[str] = field(default_factory=list)
+    warnings: list[str] = field(default_factory=list)
+    ast: exp.Expression | None = None
+
+
+def validate_sql(
+    sql: str,
+    physical: PhysicalSchema | None = None,
+    promoted_tables: set[str] | None = None,
+    forbidden_functions: set[str] | None = None,
+) -> CheckResult:
+    """Validazione statica: parsabile, un solo statement, read-only per struttura,
+    nessuna funzione in blacklist, oggetti esistenti in mschema (se fornito)."""
+    errors: list[str] = []
+    warnings: list[str] = []
+    try:
+        statements = sqlglot.parse(sql, read=DIALECT)
+    except sqlglot.errors.ParseError as e:
+        return CheckResult(ok=False, errors=[f"SQL non parsabile: {e}"])
+
+    statements = [s for s in statements if s is not None]
+    if len(statements) != 1:
+        return CheckResult(
+            ok=False,
+            errors=[f"atteso uno solo statement, trovati {len(statements)}"],
+        )
+    ast = statements[0]
+
+    # query read-only legittime: SELECT, WITH...SELECT e set operation (UNION & co.)
+    allowed_roots = (exp.Select, exp.Union, exp.Intersect, exp.Except)
+    if not isinstance(ast, allowed_roots):
+        errors.append(
+            f"solo query SELECT (incluse WITH e UNION) sono ammesse "
+            f"(trovato: {type(ast).__name__})"
+        )
+
+    for node in ast.walk():
+        for forbidden_type, label in FORBIDDEN_NODES.items():
+            if isinstance(node, forbidden_type):
+                errors.append(f"statement vietato (read-only): {label}")
+
+    blacklist = forbidden_functions if forbidden_functions is not None else DEFAULT_FORBIDDEN_FUNCTIONS
+    for func in ast.find_all(exp.Func):
+        name = (func.name or "").lower()
+        if name in blacklist:
+            errors.append(f"funzione vietata: {name}")
+
+    if physical is not None:
+        _check_objects(ast, physical, promoted_tables, errors, warnings)
+
+    return CheckResult(ok=not errors, errors=errors, warnings=warnings, ast=ast)
+
+
+def _check_objects(
+    ast: exp.Expression,
+    physical: PhysicalSchema,
+    promoted_tables: set[str] | None,
+    errors: list[str],
+    warnings: list[str],
+) -> None:
+    """Tabelle citate: devono esistere (escluse le CTE definite nella query).
+    Colonne qualificate su tabelle reali (o loro alias): devono esistere.
+    Colonne non qualificate o su CTE: non verificabili staticamente, si saltano."""
+    cte_names = {cte.alias_or_name for cte in ast.find_all(exp.CTE)}
+    alias_to_table: dict[str, str] = {}
+    for table in ast.find_all(exp.Table):
+        name = table.name
+        if name in cte_names:
+            continue
+        if name not in physical.tables:
+            errors.append(f"tabella inesistente nello schema: {name}")
+            continue
+        alias_to_table[table.alias_or_name] = name
+        if promoted_tables is not None and name not in promoted_tables:
+            warnings.append(f"tabella fuori dal perimetro promosso: {name}")
+
+    for column in ast.find_all(exp.Column):
+        qualifier = column.table
+        if not qualifier or qualifier not in alias_to_table:
+            continue
+        table_name = alias_to_table[qualifier]
+        if column.name not in physical.tables[table_name].columns:
+            errors.append(f"colonna inesistente: {table_name}.{column.name}")
diff --git a/harness/tht/vendor/VENDORED.md b/harness/tht/vendor/VENDORED.md
new file mode 100644
index 00000000..7c04456b
--- /dev/null
+++ b/harness/tht/vendor/VENDORED.md
@@ -0,0 +1,30 @@
+# Codice vendorizzato
+
+## thoth_lsh.py
+
+- Origine: `thoth_sqldb2` (pacchetto PyPI `thoth-dbmanager` 0.7.4), file `thoth_dbmanager/lsh/core.py`
+- Copyright 2025 Marco Pancotti — Apache License 2.0
+- Modifiche locali:
+  1. rimosso blocco di debug su colonna "doctype" in `create_lsh_index`;
+  2. `skip_column`: soglie parametrizzate (`max_total_chars`, `max_avg_length`) con default
+     identici agli originali (50000 / 20);
+  3. rimossa `query_lsh_index` (sostituita da `tht.lshindex.query_index`, che restituisce
+     anche gli score per il ranking spiegabile);
+  4. try/except generico rimosso da `create_lsh_index`: gli errori devono emergere.
+  5. `skip_column`: la guardia "name-like" (originariamente solo `"name"`) è stata estesa
+     a token bilingui EN/IT tramite la costante `NAME_LIKE_TOKENS`
+     (`name`, `nome`, `nominativ`, `denominazion`, `ragione_sociale`), sovrascrivibile col
+     parametro `name_tokens`. Necessario perché il datawarehouse Chirone è in italiano e le
+     colonne `nome`/`cognome`/`denominazione`/`ragione_sociale` non venivano riconosciute
+     dalla sola parola inglese `name`.
+
+**Nota: `skip_column` e `NAME_LIKE_TOKENS` non sono più usati dal flusso Thoth.**
+La selezione delle colonne da indicizzare è ora governata dal *principio di eleggibilità
+delle colonne* (spec: `docs/superpowers/specs/2026-06-13-tht-column-eligibility-principle.md`).
+L'esclusione dei testi larghi avviene a monte tramite il flag `eligible` persistito in
+`physical.yaml` (impostato da `nsp schema introspect`), non tramite l'euristica sulla
+lunghezza del vendorizzato. Le funzioni restano nel file per fedeltà alla sorgente upstream;
+Thoth ha semplicemente smesso di importarle (vedi `tht/db/sampling.py`).
+
+Le query di introspezione in `tht/db/introspect.py` sono ADATTATE (non copiate verbatim)
+da `thoth_dbmanager/adapters/postgresql.py` della stessa versione.
diff --git a/harness/tht/vendor/__init__.py b/harness/tht/vendor/__init__.py
new file mode 100644
index 00000000..e69de29b
diff --git a/harness/tht/vendor/thoth_lsh.py b/harness/tht/vendor/thoth_lsh.py
new file mode 100644
index 00000000..3007f3c3
--- /dev/null
+++ b/harness/tht/vendor/thoth_lsh.py
@@ -0,0 +1,82 @@
+# Vendored from thoth_sqldb2 (thoth-dbmanager 0.7.4) — lsh/core.py
+# Copyright 2025 Marco Pancotti — Apache License 2.0
+# Modifiche locali documentate in VENDORED.md.
+"""Core LSH (MinHash) per la ricerca di valori simili nei campi del database."""
+
+import logging
+from typing import Dict, List, Tuple
+
+from datasketch import MinHash, MinHashLSH
+from tqdm import tqdm
+
+
+def create_minhash(signature_size: int, string: str, n_gram: int) -> MinHash:
+    m = MinHash(num_perm=signature_size)
+    for d in [string[i : i + n_gram] for i in range(len(string) - n_gram + 1)]:
+        m.update(d.encode("utf8"))
+    return m
+
+
+# Token che, se presenti nel nome di una colonna, la marcano come "name-like":
+# nomi propri di persone/enti, i cui valori sono utili al value-matching e quindi
+# mai da escludere dall'indice LSH. Bilingue EN/IT (modifica locale, vedi VENDORED.md).
+# Per substring coprono le forme flesse: "nome" -> cognome/soprannome,
+# "name" -> surname/username, "denominazion" -> denominazione, ecc.
+NAME_LIKE_TOKENS: tuple[str, ...] = (
+    "name",
+    "nome",
+    "nominativ",
+    "denominazion",
+    "ragione_sociale",
+)
+
+
+def skip_column(
+    column_name: str,
+    column_values: List[str],
+    max_total_chars: int = 50000,
+    max_avg_length: int = 20,
+    name_tokens: tuple[str, ...] = NAME_LIKE_TOKENS,
+) -> bool:
+    lowered = column_name.lower()
+    if any(token in lowered for token in name_tokens):
+        return False
+    sum_of_lengths = sum(len(value) for value in column_values)
+    average_length = sum_of_lengths / len(column_values)
+    return (sum_of_lengths > max_total_chars) and (average_length > max_avg_length)
+
+
+def jaccard_similarity(m1: MinHash, m2: MinHash) -> float:
+    return m1.jaccard(m2)
+
+
+def create_lsh_index(
+    unique_values: Dict[str, Dict[str, List[str]]],
+    signature_size: int,
+    n_gram: int,
+    threshold: float,
+    verbose: bool = True,
+) -> Tuple[MinHashLSH, Dict[str, Tuple[MinHash, str, str, str]]]:
+    lsh = MinHashLSH(threshold=threshold, num_perm=signature_size)
+    minhashes: Dict[str, Tuple[MinHash, str, str, str]] = {}
+    total = sum(
+        len(column_values)
+        for table_values in unique_values.values()
+        for column_values in table_values.values()
+    )
+    logging.info("Total unique values: %s", total)
+    progress_bar = tqdm(total=total, desc="Creating LSH") if verbose else None
+
+    for table_name, table_values in unique_values.items():
+        for column_name, column_values in table_values.items():
+            for idx, value in enumerate(column_values):
+                minhash = create_minhash(signature_size, value, n_gram)
+                minhash_key = f"{table_name}_{column_name}_{idx}"
+                minhashes[minhash_key] = (minhash, table_name, column_name, value)
+                lsh.insert(minhash_key, minhash)
+                if progress_bar:
+                    progress_bar.update(1)
+
+    if progress_bar:
+        progress_bar.close()
+    return lsh, minhashes