From 04b9ed1773a5bcbf86b6ba690bd65079a8c314a0 Mon Sep 17 00:00:00 2001
From: Zio Gabber <78922322+Gabrymi93@users.noreply.github.com>
Date: Tue, 20 Oct 2026 21:53:04 +0100
Subject: [PATCH] fix(registry): registra i dataset mart-only nel catalogo

I compose senza clean: non entravano nella sezione datasets: i mart restavano orfani e ACB/MCP non vedevano il dataset. Ora build_clean_catalog emette una entry con metadati e mart_refs, senza columns/location; lo schema rende i due campi opzionali.
---
 CHANGELOG.md                                  |  4 +++
 tests/test_registry_builder.py                | 27 ++++++++++++++++---
 toolkit/registry/builders.py                  | 19 ++++++++-----
 toolkit/registry/schemas/registry.schema.json | 10 +++----
 4 files changed, 45 insertions(+), 15 deletions(-)

diff --git a/CHANGELOG.md b/CHANGELOG.md
index cba9a9c..3632a96 100644
--- a/CHANGELOG.md
+++ b/CHANGELOG.md
@@ -1,5 +1,9 @@
 ## [Unreleased]
 
+### Added
+
+- **Registry: dataset mart-only**: i compose senza `clean:` non restano più fuori dal catalogo con i mart orfani. `build_clean_catalog` emette una entry dataset con metadati e `mart_refs`, senza `columns`/`location`; `registry.schema.json` rende i due campi opzionali. Così ACB/MCP vedono il dataset (`find`/`explore`) e risolvono i suoi mart (`gcs_query`).
+
 ### Changed
 
 - **Layering ripristinato** (PR #504): rimosse 6 inversioni di dipendenza `core → clean/domain/scout/scaffold`. `sql_validation` spostato in `domain/`, orchestration support in `domain/support.py`, `ProbePool` in `scout/pool.py`, scaffold clean.sql spostato da `raw/run.py` a `cmd_run` (comportamento invariato). Contratto condiviso `dataset_yml_slug`/`iter_workspace_dataset_ymls` in `registry/layout.py`.
diff --git a/tests/test_registry_builder.py b/tests/test_registry_builder.py
index a36e325..f34cc90 100644
--- a/tests/test_registry_builder.py
+++ b/tests/test_registry_builder.py
@@ -21,6 +21,7 @@
     build_clean_catalog,
     build_entity_graph,
     build_mart_catalog,
+    build_registry,
     build_signals,
 )
 from toolkit.registry.layout import RepoLayout
@@ -64,6 +65,8 @@
   tags: [test]
   category: demo
   years: [2024]
+clean:
+  sql: "sql/clean.sql"
 """
 
 RUN_RECORD = {
@@ -184,8 +187,8 @@ def test_clean_catalog_dirname_not_identity(tmp_path: Path) -> None:
 
 @pytest.mark.pure_unit
 def test_clean_catalog_missing_parquet_reported(tmp_path: Path) -> None:
-    """Dataset dichiarato senza parquet locale → errore, non catalogo finto."""
-    layout = _make_repo(tmp_path)
+    """Dataset con clean.sql ma senza parquet locale → errore, non catalogo finto."""
+    layout = _make_repo(tmp_path, with_mart=False)
     import shutil
 
     shutil.rmtree(layout.repo_root / "out" / "data" / "clean")
@@ -194,6 +197,24 @@ def test_clean_catalog_missing_parquet_reported(tmp_path: Path) -> None:
     assert catalog["datasets"] == []
 
 
+@pytest.mark.contract
+def test_clean_catalog_registers_mart_only_dataset(tmp_path: Path) -> None:
+    """Mart-only (no clean.sql né parquet): dataset con mart_refs, senza columns/location."""
+    layout = _make_repo(tmp_path)  # MART_YML: nessun clean.sql
+    import shutil
+
+    shutil.rmtree(layout.repo_root / "out" / "data" / "clean")
+    built = build_registry(layout)
+
+    assert built["errors"]["registry"]["validation"] == []
+    ds = built["registry"]["datasets"][0]
+    assert ds["slug"] == "my_dataset"
+    assert ds["period"] == {"start": 2000, "end": 2024}
+    assert ds["mart_refs"] == ["my_dataset__mart_trend"]
+    assert "columns" not in ds
+    assert "location" not in ds
+
+
 @pytest.mark.contract
 def test_clean_catalog_slims_editorial_entry_without_parquet(tmp_path: Path) -> None:
     """Entry editoriale (no parquet locale) preservata MA filtrata dalle chiavi morte.
@@ -204,7 +225,7 @@ def test_clean_catalog_slims_editorial_entry_without_parquet(tmp_path: Path) ->
     (_slim_dataset_entry) deve comunque applicarsi, così come il fix role
     (year → dimension).
     """
-    layout = _make_repo(tmp_path)
+    layout = _make_repo(tmp_path, with_mart=False)
     import shutil
 
     shutil.rmtree(layout.repo_root / "out" / "data" / "clean")
diff --git a/toolkit/registry/builders.py b/toolkit/registry/builders.py
index b6aec27..262b938 100644
--- a/toolkit/registry/builders.py
+++ b/toolkit/registry/builders.py
@@ -144,7 +144,7 @@ def build_clean_catalog(
             continue
 
         columns, latest_year = latest_clean_columns(manifest, alias_map)
-        if columns is None:
+        if columns is None and not manifest.is_mart_only:
             # Nessun parquet locale: possibile solo se l'entry è editoriale
             # (preservata) — altrimenti errore e skip.
             if manifest.slug in editorial:
@@ -155,12 +155,16 @@ def build_clean_catalog(
                 f"per anni {manifest.years} (runnare la pipeline per generarlo)"
             )
             continue
-        if latest_year is None:
+        if latest_year is None and not manifest.is_mart_only:
             derive_errors.append(f"{manifest.slug}: schema letto ma anno non risolvibile")
             continue
 
         years_present = sorted(y for y in manifest.years if clean_parquet_path(manifest.cfg, y))
-        years_eff = years_present or [latest_year]
+        # mart-only: nessun clean locale → period dagli years dichiarati
+        years_eff = years_present or ([latest_year] if latest_year else list(manifest.years))
+        if not years_eff:
+            derive_errors.append(f"{manifest.slug}: years non dichiarati — period non risolvibile")
+            continue
 
         entry: dict[str, Any] = {
             "slug": manifest.slug,
@@ -171,10 +175,13 @@ def build_clean_catalog(
             "period": manifest.period or {"start": years_eff[0], "end": years_eff[-1]},
             "tags": _dedup_tags(manifest.tags),
             "category": manifest.category,
-            "columns": columns,
-            "location": contract.clean_location(manifest.slug, years_eff),
             "stage": "incubating",
         }
+        # Un dataset mart-only non ha clean: niente columns/location. La card
+        # resta comunque visibile e i mart sono linkati da mart_refs.
+        if columns is not None:
+            entry["columns"] = columns
+            entry["location"] = contract.clean_location(manifest.slug, years_eff)
 
         # Merge editoriale: i campi umani sovrascrivono. role è SEMPRE
         # derivato (semantic_type/type), mai dall'editoriale: l'existing
@@ -194,7 +201,7 @@ def build_clean_catalog(
             if not manifest.period and old.get("period"):
                 entry["period"] = old["period"]
             old_cols = {c.get("name"): c for c in old.get("columns", [])}
-            for col in entry["columns"]:
+            for col in entry.get("columns", []):
                 oc = old_cols.get(col["name"])
                 if oc:
                     if oc.get("description"):
diff --git a/toolkit/registry/schemas/registry.schema.json b/toolkit/registry/schemas/registry.schema.json
index 8d052ea..8a10fb0 100644
--- a/toolkit/registry/schemas/registry.schema.json
+++ b/toolkit/registry/schemas/registry.schema.json
@@ -29,7 +29,7 @@
       "items": {
         "$ref": "#/$defs/dataset"
       },
-      "description": "Inventario queryabile dei clean parquet (ex clean_catalog)."
+      "description": "Dataset del repo (ex clean_catalog). I dataset mart-only non hanno clean: columns e location sono opzionali."
     },
     "marts": {
       "type": "array",
@@ -73,9 +73,7 @@
         "slug",
         "name",
         "description",
-        "period",
-        "columns",
-        "location"
+        "period"
       ],
       "properties": {
         "slug": {
@@ -141,10 +139,10 @@
         },
         "columns": {
           "type": "array",
-          "minItems": 1,
           "items": {
             "$ref": "#/$defs/column"
-          }
+          },
+          "description": "Colonne del clean. Assente/vuoto per i dataset mart-only."
         },
         "location": {
           "$ref": "#/$defs/location"