feat(catalog): tokenized semantic find + route free-text discovery to ACB

tests/test_catalog_ops.py

@@ -734,6 +734,45 @@ def test_find_by_column(self, resolver_semantic: CatalogResolver) -> None:
             {"name": "importo", "type": "DOUBLE", "role": "metric", "description": "Importo gara"}
         ]
 
+    def test_find_multi_token_cross_meta(self, resolver_semantic: CatalogResolver) -> None:
+        """Regression: i token di una query naturale possono spandersi tra
+        campi meta (AND cross-part) — prima serviva la frase intera continua."""
+        # 'appalti' solo in tags/category, 'gara' in name/description
+        res = resolver_semantic.list_datasets(query="appalti gara")
+        slugs = {d["slug"] for d in res["datasets"]}
+        assert "anac_bandi_gara" in slugs
+        assert "terna_electricity_by_source" not in slugs
+
+    def test_find_drops_stopwords(self, resolver_semantic: CatalogResolver) -> None:
+        """Le stopword italiane non sono token richiesti."""
+        res = resolver_semantic.list_datasets(query="bandi di gara")
+        slugs = {d["slug"] for d in res["datasets"]}
+        assert "anac_bandi_gara" in slugs
+        # solo stopword → nessun hit semantico
+        res2 = resolver_semantic.list_datasets(query="di dei delle")
+        assert res2["datasets"] == []
+
+    def test_find_morphology_plural(self, resolver_semantic: CatalogResolver) -> None:
+        """'gare' matcha 'gara' (final-vowel strip)."""
+        res = resolver_semantic.list_datasets(query="gare appalti")
+        slugs = {d["slug"] for d in res["datasets"]}
+        assert "anac_bandi_gara" in slugs
+
+    def test_find_column_prefix(self, resolver_semantic: CatalogResolver) -> None:
+        """Prefisso di nome colonna: 'import' → 'importo'."""
+        res = resolver_semantic.list_datasets(query="import")
+        entry = next(d for d in res["datasets"] if d["slug"] == "anac_bandi_gara")
+        assert entry["meta_match"] is not True
+        assert [c["name"] for c in entry["matched_columns"]] == ["importo"]
+
+    def test_token_in_text_no_substring_false_positive(self) -> None:
+        """'appalti' non matcha dentro 'subappalti' (split per parola)."""
+        from lab_connectors.textmatch import token_in_text
+
+        assert not token_in_text("appalti", "n_subappalti")
+        assert token_in_text("appalti", "appalti pubblici")
+        assert token_in_text("gara", "gare")  # morphologia
+
     def test_find_metric_only(self, resolver_semantic: CatalogResolver) -> None:
         """metric_only esclude i dataset senza colonne metric."""
         res = resolver_semantic.list_datasets(metric_only=True)

toolkit/domain/catalog.py

@@ -21,6 +21,7 @@
 from typing import Any, Literal
 
 from lab_connectors.gcs.paths import CLEAN_BUCKET, MART_BUCKET
+from lab_connectors.textmatch import query_tokens, token_in_text
 
 from toolkit.core.duckdb_shape import parquet_preview
 from toolkit.core.io import read_json_or_none, read_yaml
@@ -439,24 +440,30 @@ def _semantic_query_hits(
 ) -> dict[str, list[dict[str, Any] | None]]:
     """Slug che matchano la query sulla semantica (meta o colonne).
 
+    Tokenizza la query (stopword italiane escluse, primitive condivise in
+    ``lab_connectors.textmatch``) e richiede che **tutti** i token matchino
+    sparsi tra slug + meta (AND cross-part), oppure tutti dentro una singola
+    colonna. Prima serviva la frase intera come substring continua:
+    "appalti pubblici anac" non matchava mai.
+
     Returns:
         ``{slug: [None | colonna_matchata, ...]}`` — ``None`` = meta match
         (name/description/source/tags/category), dict = colonna matchata.
     """
-    q = query.lower()
+    tokens = query_tokens(query)
+    if not tokens:
+        return {}
     hits: dict[str, list[dict[str, Any] | None]] = {}
     for slug, s in semantic.items():
         matched: list[dict[str, Any] | None] = []
         meta_parts = " ".join(
             str(s.get(k) or "") for k in ("name", "description", "source", "tags", "category")
         )
-        if q in slug.lower() or q in meta_parts.lower():
+        if all(token_in_text(token, slug) or token_in_text(token, meta_parts) for token in tokens):
             matched.append(None)
         for col in s.get("columns", []) or []:
-            if (
-                q in str(col.get("name", "")).lower()
-                or q in str(col.get("description", "")).lower()
-            ):
+            col_text = f"{col.get('name', '')} {col.get('description', '')}"
+            if all(token_in_text(token, col_text) for token in tokens):
                 matched.append(
                     {
                         "name": col.get("name"),

toolkit/mcp/server.py

@@ -74,7 +74,10 @@
     description=(
         "Ispezione dataset: find, overview, status, preflight, schema-diff, related.\n\n"
         "Actions:\n"
-        "- find: cerca dataset per slug/testo/source (params: query, layer, limit, source, stage, status_filter)\n"
+        "- find: cerca dataset per slug, colonne o meta (params: query, layer, limit, "
+        "source, stage, status_filter). Per discovery org-wide con query naturali "
+        "usa dataciviclab-context find; toolkit find è lo strumento live su "
+        "workspace+GCS e cerca anche per nome/descrizione colonna.\n"
         "- overview: schema colonne + conteggio + preview (params: slug, layer, year, source, profile)\n"
         "- status: stato completo dataset (params: config_path, year, since, until)\n"
         "- preflight: diagnostica pre-run (params: config_path, years)\n"