diff --git a/scripts/collectors/_validate_base.py b/scripts/collectors/_validate_base.py
index 465b291..1ddc94d 100644
--- a/scripts/collectors/_validate_base.py
+++ b/scripts/collectors/_validate_base.py
@@ -50,8 +50,25 @@ def pick_best_url(items: list[dict[str, Any]]) -> dict[str, Any] | None:
 # ── Reachability probe ────────────────────────────────────────────────────────
 
 
-def probe_reachability(url: str, timeout: int = _DEFAULT_TIMEOUT) -> dict[str, Any]:
-    """HTTP HEAD probe, con fallback SSL."""
+# Soglia circuit breaker: 3 fallimenti consecutivi sullo stesso host e il
+# client smette di tentare la rete (CircuitOpenError immediato). Serve per
+# non bruciare retry+timeout su host giu' ripetuti gruppo dopo gruppo
+# (es. lavoro_opendata: 83 gruppi x ~11s = ~15min senza breaker).
+CIRCUIT_THRESHOLD = 3
+
+
+def probe_reachability(
+    url: str,
+    timeout: int = _DEFAULT_TIMEOUT,
+    client: Any | None = None,
+) -> dict[str, Any]:
+    """HTTP HEAD probe, con fallback SSL.
+
+    ``client`` opzionale: se fornito (es. dal validate che condivide un
+    HttpClient con circuit breaker), viene riusato — il breaker resta
+    attivo tra gruppi della stessa fonte. Altrimenti crea un client nuovo
+    (comportamento storico).
+    """
     result: dict[str, Any] = {
         "reachable": False,
         "status_code": None,
@@ -59,7 +76,7 @@ def probe_reachability(url: str, timeout: int = _DEFAULT_TIMEOUT) -> dict[str, A
         "error": None,
     }
     try:
-        client = HttpClient(timeout=timeout)
+        client = client or HttpClient(timeout=timeout)
         resp = client.head(url, verify=False)
         if resp.is_ok and resp.response is not None:
             result["reachable"] = True
@@ -85,9 +102,170 @@ def _is_csv_url(url: str, content_type: str | None = None) -> bool:
     return url.lower().endswith(".csv") or ".csv?" in url.lower()
 
 
-def sniff_csv_schema(url: str, timeout: int = _DEFAULT_TIMEOUT) -> dict[str, Any]:
+# Nomi colonna che indicano una dimensione temporale (valori = anni)
+_YEAR_COLUMN_HINTS = frozenset(
+    {
+        "anno",
+        "year",
+        "annual",
+        "data",
+        "date",
+        "periodo",
+        "period",
+        "mese",
+        "month",
+        "time_period",
+        "anno di riferimento",
+        "anno di liquidazione",
+        "anno tariffario",
+        "anno di competenza",
+    }
+)
+
+
+def _is_year_column(col: str) -> bool:
+    """True se il nome colonna indica una dimensione temporale (anni)."""
+    cl = col.lower().strip()
+    return cl in _YEAR_COLUMN_HINTS or cl.startswith("anno") or "time_period" in cl
+
+
+def _extract_year_range(
+    raw: bytes | None,
+    columns: list[str],
+    sample: list[dict],
+    url: str,
+) -> tuple[int | None, int | None]:
+    """Estrae range anni (min, max) da un CSV profilato.
+
+    Ordine di tentativo:
+    1. anni nei NOMI colonna (es. ``anno_2020``) — toolkit._extract_years_from_columns
+    2. valori in una colonna-anno (es. colonna ``ANNO`` con valori 2018..2024):
+       il sample dello sniff standard contiene stringhe, quindi serve DuckDB
+       sul raw per avere valori tipizzati e range su tutto il sample
+    3. anni nel filename (es. ``beneficiari_2007-2013.zip``) — toolkit._extract_years
+
+    Ritorna (None, None) se non trova nulla.
+    """
+    from toolkit.profile.preview import _extract_years_from_columns
+    from toolkit.scout.link_extractor import _extract_years as _filename_years
+
+    # 1. anni da nomi colonna
+    y_cols = _extract_years_from_columns(columns)
+    if y_cols != (None, None):
+        return y_cols
+
+    # 2. anni da valori in colonna-anno (serve sample tipizzato via DuckDB)
+    if raw and any(_is_year_column(col) for col in columns):
+        duck = _sniff_csv_duckdb(raw, _SNIFF_BYTES)
+        if duck.get("year_min") is not None:
+            return (duck["year_min"], duck["year_max"])
+
+    # 3. anni da filename
+    y_file = _filename_years(url.split("?")[0])
+    if y_file:
+        return (min(y_file), max(y_file))
+
+    return (None, None)
+
+
+def _sniff_csv_duckdb(raw: bytes, sample_size: int) -> dict[str, Any]:
+    """Fallback sniff via DuckDB read_csv.
+
+    Il parser standard (csv.Sniffer + DictReader) fallisce su SDMX-CSV e su
+    CSV con newline dentro campi non quotati ("new-line character seen in
+    unquoted field") — DuckDB li gestisce. Usato solo quando il tentativo
+    standard non produce colonne.
+    """
+    import tempfile
+
+    import duckdb
+
+    result: dict[str, Any] = {
+        "columns": [],
+        "num_columns": 0,
+        "num_rows": 0,
+        "delimiter": None,
+        "encoding": "utf-8",
+        "sample": [],
+        "error": None,
+    }
+    try:
+        with tempfile.NamedTemporaryFile(suffix=".csv", delete=False) as tmp:
+            tmp.write(raw)
+            tmp_path = tmp.name
+        try:
+            df = duckdb.sql(
+                f"SELECT * FROM read_csv('{tmp_path}', header=true, sample_size=-1)"
+            ).fetchdf()
+            result["columns"] = list(df.columns)
+            result["num_columns"] = len(df.columns)
+            result["num_rows"] = len(df)
+            result["delimiter"] = None  # DuckDB auto-detect; non esposto
+            result["sample"] = df.head(5).to_dict("records")[:5]
+            # Range anni dalla colonna-anno sull'intero sample (non solo 5 righe)
+            ymin, ymax = _year_range_from_df(df)
+            if ymin is not None:
+                result["year_min"] = ymin
+                result["year_max"] = ymax
+        finally:
+            import os
+
+            os.unlink(tmp_path)
+    except Exception as exc:
+        result["error"] = f"duckdb fallback: {exc}"
+    return result
+
+
+def _year_range_from_df(df) -> tuple[int | None, int | None]:
+    """Min/max anni dai valori della colonna-anno in un DataFrame DuckDB."""
+    import math
+
+    if df is None or len(df.columns) == 0:
+        return (None, None)
+    # trova la colonna-anno (nome hint o valori che sembrano anni)
+    year_col: str | None = None
+    for col in df.columns:
+        cl = str(col).lower().strip()
+        if cl in _YEAR_COLUMN_HINTS or cl.startswith("anno") or "time_period" in cl:
+            year_col = col
+            break
+    if year_col is None:
+        # fallback: colonna numerica con 2+ valori nel range 1900-2100
+        for col in df.columns:
+            vals = [
+                int(v)
+                for v in df[col].tolist()
+                if isinstance(v, (int, float))
+                and not (isinstance(v, float) and math.isnan(v))
+                and 1900 <= v <= 2100
+            ]
+            if len(set(vals)) >= 2:
+                year_col = col
+                break
+    if year_col is None:
+        return (None, None)
+    vals = [
+        int(v)
+        for v in df[year_col].tolist()
+        if isinstance(v, (int, float))
+        and not (isinstance(v, float) and math.isnan(v))
+        and 1900 <= v <= 2100
+    ]
+    if not vals:
+        return (None, None)
+    return (min(vals), max(vals))
+
+
+def sniff_csv_schema(
+    url: str,
+    timeout: int = _DEFAULT_TIMEOUT,
+    client: Any | None = None,
+) -> dict[str, Any]:
     """Download primi byte CSV e sniffa schema.
 
+    ``client`` opzionale: se fornito, riusato (breaker condiviso tra
+    gruppi). Altrimenti client nuovo (comportamento storico).
+
     Returns:
         columns, num_columns, num_rows, delimiter, encoding, sample, error
     """
@@ -101,7 +279,7 @@ def sniff_csv_schema(url: str, timeout: int = _DEFAULT_TIMEOUT) -> dict[str, Any
         "error": None,
     }
     try:
-        client = HttpClient(timeout=timeout)
+        client = client or HttpClient(timeout=timeout)
         # Range limitato: scarica solo primi KB (SDMX puo' essere enorme)
         resp = client.get(url, headers={"Range": f"bytes=0-{_SNIFF_BYTES}"}, verify=False)
         if not resp.is_ok or resp.response is None:
@@ -144,6 +322,34 @@ def sniff_csv_schema(url: str, timeout: int = _DEFAULT_TIMEOUT) -> dict[str, Any
         result["delimiter"] = delimiter
         result["encoding"] = encoding
 
+        # Fallback DuckDB: il parser standard fallisce su SDMX-CSV e CSV con
+        # newline nei campi ("new-line character seen in unquoted field").
+        # Se non ha prodotto colonne, prova DuckDB che li gestisce.
+        if not columns:
+            duck = _sniff_csv_duckdb(raw, _SNIFF_BYTES)
+            if duck.get("columns"):
+                result["columns"] = duck["columns"]
+                result["num_columns"] = duck["num_columns"]
+                result["num_rows"] = duck["num_rows"]
+                result["sample"] = duck["sample"]
+                result["sniff_fallback"] = "duckdb"
+                result["error"] = None  # recuperato: errore standard superato
+            elif duck.get("error"):
+                result["sniff_error"] = duck["error"]
+
+        # Estrazione range anni (min, max) da colonne/valori/filename.
+        # Popola year_min/year_max quando la fonte non li fornisce gia'.
+        if result.get("columns"):
+            ymin, ymax = _extract_year_range(
+                raw=raw,
+                columns=result["columns"],
+                sample=result.get("sample") or [],
+                url=url,
+            )
+            if ymin is not None and ymax is not None:
+                result["year_min"] = ymin
+                result["year_max"] = ymax
+
     except Exception as e:
         result["error"] = str(e)
 
@@ -156,6 +362,7 @@ def sniff_csv_schema(url: str, timeout: int = _DEFAULT_TIMEOUT) -> dict[str, Any
 def validate_tabular_group(
     items: list[dict[str, Any]],
     deep: bool = False,
+    client: Any | None = None,
 ) -> dict[str, Any]:
     """Valida un gruppo di item tabulari (CKAN/HTML): HEAD + sniff CSV.
 
@@ -166,6 +373,11 @@ def validate_tabular_group(
     approfondito (tipi colonna, quality score, granularità, anni) — più
     lento ma più ricco.
 
+    ``client`` opzionale: HttpClient condiviso tra gruppi (con circuit
+    breaker attivo) — se fornito dalla pipeline, il breaker sopravvive
+    tra gruppi della stessa run. Altrimenti ne crea uno per gruppo
+    (comportamento storico, breaker non efficace tra gruppi).
+
     Usato da collectors.ckan.validate_items() e collectors.html.validate_items().
     """
     best = pick_best_url(items)
@@ -202,9 +414,11 @@ def validate_tabular_group(
     }
 
     # Propaga metadati del gruppo (dal merge)
+    # Attenzione: il merge pandas produce NaN, non None — tratta entrambi come "assente".
     for col in ("dataset_group_size", "dataset_group_year_min", "dataset_group_year_max"):
         val = best.get(col)
-        if val is not None:
+        is_nan = isinstance(val, float) and val != val  # NaN
+        if val is not None and not is_nan:
             result[col] = val
 
     # ── Non-CSV: salta completamente i probe HTTP ──────────────────────────
@@ -217,7 +431,14 @@ def validate_tabular_group(
         return result
 
     # ── CSV: HEAD probe + sniff schema ─────────────────────────────────────
-    probe = probe_reachability(url)
+    # Client condiviso con circuit breaker: se un host fallisce 3 volte
+    # consecutive, i gruppi successivi della stessa fonte non ritentano la
+    # rete (CircuitOpenError immediato) — evita di bruciare ~11s per gruppo
+    # su host giu' (es. lavoro_opendata: 83 gruppi -> ~15min senza breaker).
+    # Se la pipeline passa un client condiviso, il breaker sopravvive tra
+    # gruppi; altrimenti ne crea uno per gruppo (breaker solo intra-gruppo).
+    http = client or HttpClient(timeout=_DEFAULT_TIMEOUT, circuit_threshold=CIRCUIT_THRESHOLD)
+    probe = probe_reachability(url, timeout=_DEFAULT_TIMEOUT, client=http)
     result.update({k: v for k, v in probe.items()})
 
     if not result["reachable"]:
@@ -225,7 +446,7 @@ def validate_tabular_group(
         return result
 
     # Sniff CSV: sniff leggero interno.
-    schema = sniff_csv_schema(url)
+    schema = sniff_csv_schema(url, timeout=_DEFAULT_TIMEOUT, client=http)
     result["columns"] = schema["columns"]
     result["num_columns"] = schema["num_columns"]
     result["num_sample_rows"] = schema["num_rows"]
@@ -234,6 +455,17 @@ def validate_tabular_group(
     if schema["error"]:
         result["sniff_error"] = schema["error"]
 
+    # Anni: se il gruppo non li forniva gia' (year_signal da collector),
+    # usa il range estratto dallo sniff (colonne/valori/filename).
+    # Attenzione: il merge pandas produce NaN, non None — tratta entrambi come "assente".
+    ymin_cur = result.get("dataset_group_year_min")
+    missing = ymin_cur is None or (
+        isinstance(ymin_cur, float) and ymin_cur != ymin_cur  # NaN
+    )
+    if missing and schema.get("year_min") is not None:
+        result["dataset_group_year_min"] = schema["year_min"]
+        result["dataset_group_year_max"] = schema["year_max"]
+
     # readiness_score 0-10
     score = 0
     if result.get("reachable"):
diff --git a/scripts/collectors/ckan.py b/scripts/collectors/ckan.py
index 1dccc76..2d46140 100644
--- a/scripts/collectors/ckan.py
+++ b/scripts/collectors/ckan.py
@@ -631,11 +631,16 @@ def _ckan_standard_path(
     }
 
 
-def validate_items(items: list[dict[str, Any]]) -> dict[str, Any]:
+def validate_items(
+    items: list[dict[str, Any]],
+    client: Any | None = None,
+) -> dict[str, Any]:
     """Valida un gruppo di item CKAN.
 
     Usa il validatore tabulare standard: pick_best_url → HEAD → sniff CSV.
+    ``client`` opzionale: HttpClient condiviso con circuit breaker, passato
+    al validatore tabulare per mantenere lo stato breaker tra gruppi.
     """
     from ._validate_base import validate_tabular_group
 
-    return validate_tabular_group(items)
+    return validate_tabular_group(items, client=client)
diff --git a/scripts/collectors/sdmx.py b/scripts/collectors/sdmx.py
index a5dfb7e..fe1bf60 100644
--- a/scripts/collectors/sdmx.py
+++ b/scripts/collectors/sdmx.py
@@ -97,12 +97,17 @@ def collect(source_id: str, source_cfg: dict[str, Any], captured_at: str) -> Col
     return CollectorResult(rows=rows)
 
 
-def validate_items(items: list[dict[str, Any]]) -> dict[str, Any]:
+def validate_items(
+    items: list[dict[str, Any]],
+    client: Any | None = None,  # noqa: ARG001 — SDMX non fa probe HTTP
+) -> dict[str, Any]:
     """Valida un gruppo di item SDMX.
 
     SDMX non supporta HEAD/HTTP probe. La validazione si basa
     sui metadati gia' raccolti dal collector durante l'inventory:
     se ha api_base_url e distribution_url → raggiungibile.
+    ``client`` accettato per firma uniforme col validatore tabulare,
+    ma ignorato (SDMX non fa HTTP qui).
     """
     if not items:
         return {
diff --git a/scripts/collectors/sparql.py b/scripts/collectors/sparql.py
index 82268ba..c6fa23c 100644
--- a/scripts/collectors/sparql.py
+++ b/scripts/collectors/sparql.py
@@ -82,10 +82,15 @@ def collect(source_id: str, source_cfg: dict, captured_at: str) -> CollectorResu
     return CollectorResult(rows=rows)
 
 
-def validate_items(items: list[dict]) -> dict[str, Any]:
+def validate_items(
+    items: list[dict],
+    client: Any | None = None,  # noqa: ARG001 — SPARQL non riusa il client HTTP tabulare
+) -> dict[str, Any]:
     """Validate a group of SPARQL items.
 
     Checks endpoint reachability and runs a simple COUNT query.
+    ``client`` accettato per firma uniforme col validatore tabulare,
+    ma ignorato (SPARQL usa il proprio path HTTP).
     """
     if not items:
         return {
diff --git a/scripts/pipeline/run_pipeline.py b/scripts/pipeline/run_pipeline.py
index b970308..3518446 100644
--- a/scripts/pipeline/run_pipeline.py
+++ b/scripts/pipeline/run_pipeline.py
@@ -87,12 +87,12 @@ def run_merge(df: pd.DataFrame, dry_run: bool = False) -> pd.DataFrame:
 # ── Step 3: Validate ──────────────────────────────────────────────────────────
 
 
-def _validate_one(group_df: pd.DataFrame) -> dict:
+def _validate_one(group_df: pd.DataFrame, client=None) -> dict:
     """Validate a single group: pick protocol, dispatch validator."""
     items = group_df.to_dict("records")
     protocol = str(items[0].get("protocol", "")) if items else ""
     validator = dispatch_validate(protocol)
-    return validator(items)
+    return validator(items, client=client)
 
 
 def run_validate(
@@ -116,10 +116,20 @@ def run_validate(
     csv_ok = 0
     t0 = time.time()
 
+    # Client condiviso con circuit breaker: se un host fallisce N volte
+    # consecutive (es. lavoro_opendata giu'), i gruppi successivi della
+    # stessa fonte non ritentano la rete — CircuitOpenError immediato.
+    # Un solo client per tutta la run => il breaker e' cross-gruppo.
+    # timeout=5 coerente con _validate_base: il default del client (60s)
+    # farebbe bruciare ~114s per fallimento prima che il breaker apra.
+    from lab_connectors.http import HttpClient
+
+    http_client = HttpClient(timeout=5, circuit_threshold=3)
+
     if workers <= 1:
         # Sequenziale
         for i, (group_name, group_df) in enumerate(groups):
-            result = _validate_one(group_df)
+            result = _validate_one(group_df, client=http_client)
             results[i] = result
             if result.get("reachable"):
                 ok += 1
@@ -132,7 +142,8 @@ def run_validate(
         # Parallelo con ThreadPoolExecutor
         with ThreadPoolExecutor(max_workers=workers) as pool:
             future_map = {
-                pool.submit(_validate_one, group_df): i for i, (_, group_df) in enumerate(groups)
+                pool.submit(_validate_one, group_df, http_client): i
+                for i, (_, group_df) in enumerate(groups)
             }
             for future in as_completed(future_map):
                 i = future_map[future]
diff --git a/tests/pipeline/test_run_pipeline.py b/tests/pipeline/test_run_pipeline.py
new file mode 100644
index 0000000..c913757
--- /dev/null
+++ b/tests/pipeline/test_run_pipeline.py
@@ -0,0 +1,97 @@
+"""
+Test run_pipeline — merge/validate orchestration (con fake_http).
+
+Copre il path nuovo del client condiviso con circuit breaker:
+``_validate_one`` deve passare il client al validatore, e ``run_validate``
+deve creare un solo client per tutta la run.
+
+Marker: pure_unit (nessuna chiamata HTTP reale).
+"""
+
+from __future__ import annotations
+
+import pandas as pd
+import pytest
+from lab_connectors.http import HttpResult
+from lab_connectors.testing import fake_response
+
+pytestmark = pytest.mark.pure_unit
+
+MEF_URL = (
+    "https://www1.finanze.gov.it/finanze/analisi_stat/public/v_4_0_0/"
+    "contenuti/REG_tipo_reddito_2025.csv?d=1615465800"
+)
+MEF_URL_NO_QUERY = MEF_URL.split("?")[0]
+
+
+def _make_group_df(suffix: str = "") -> pd.DataFrame:
+    return pd.DataFrame(
+        [
+            {
+                "dataset_group": f"mef_irpef/statistiche/reg-tipo-reddito{suffix}",
+                "source_id": "mef_irpef",
+                "protocol": "ckan",
+                "distribution_url": MEF_URL,
+                "format": "csv",
+                "year_signal": 2025,
+            }
+        ]
+    )
+
+
+def test_validate_one_with_shared_client(monkeypatch, fake_http):
+    """_validate_one passa il client condiviso al validatore."""
+    from scripts.pipeline import run_pipeline
+
+    csv_body = "ANNO,REGIONE,IMPORTO\n2020,Lazio,1000\n2021,Lombardia,2000\n"
+    fake_http.responses[MEF_URL_NO_QUERY] = HttpResult(
+        response=fake_response(200, csv_body), err=None
+    )
+    # il validate crea client con HttpClient(...) — patchiamo la factory
+    # (importata da lab_connectors.http sia in _validate_base sia in run_pipeline)
+    monkeypatch.setattr("lab_connectors.http.HttpClient", lambda **kw: fake_http)
+
+    df = _make_group_df()
+    result = run_pipeline._validate_one(df, client=fake_http)
+
+    assert result["reachable"] is True
+    assert len(result.get("columns", [])) > 0
+    # il client passato è stato usato: il fake registra la richiesta
+    assert fake_http.requests, "il client condiviso deve essere stato usato"
+
+
+def test_validate_one_without_client_creates_own(monkeypatch, fake_http):
+    """Senza client, _validate_one ne crea uno (comportamento storico)."""
+    from scripts.pipeline import run_pipeline
+
+    csv_body = "ANNO,REGIONE,IMPORTO\n2020,Lazio,1000\n"
+    fake_http.responses[MEF_URL_NO_QUERY] = HttpResult(
+        response=fake_response(200, csv_body), err=None
+    )
+    # _validate_base importa HttpClient a livello modulo: patchiamo il nome
+    # bindato nel modulo consumer (il path lab_connectors.http non basta quando
+    # il pacchetto e' installato, non editable)
+    monkeypatch.setattr("scripts.collectors._validate_base.HttpClient", lambda **kw: fake_http)
+
+    df = _make_group_df()
+    result = run_pipeline._validate_one(df)
+
+    assert result["reachable"] is True
+
+
+def test_run_validate_uses_shared_client(monkeypatch, fake_http):
+    """run_validate crea UN client e lo passa a tutti i gruppi."""
+    from scripts.pipeline import run_pipeline
+
+    csv_body = "ANNO,REGIONE,IMPORTO\n2020,Lazio,1000\n"
+    fake_http.responses[MEF_URL_NO_QUERY] = HttpResult(
+        response=fake_response(200, csv_body), err=None
+    )
+    monkeypatch.setattr("lab_connectors.http.HttpClient", lambda **kw: fake_http)
+
+    # due gruppi distinti per verificare che il client sia condiviso
+    df = pd.concat([_make_group_df("A"), _make_group_df("B")], ignore_index=True)
+    results = run_pipeline.run_validate(df, max_groups=2, workers=1)
+
+    ok = [r for r in results if r is not None and r.get("reachable")]
+    assert len(ok) == 2
diff --git a/tests/pipeline/test_validate_utils.py b/tests/pipeline/test_validate_utils.py
index d67cf2d..ee793aa 100644
--- a/tests/pipeline/test_validate_utils.py
+++ b/tests/pipeline/test_validate_utils.py
@@ -1,18 +1,29 @@
 """
-Test validate utilities with real URLs from inventory.
+Test validate utilities — probe, sniff, anni, cluster.
+
+I test di rete usano ``fake_http`` (FakeHttpClient da lab_connectors.testing)
+per essere deterministici e non dipendere da fonti esterne. I test smoke
+(su URL reali) sono in ``TestSmokeNetwork``: non fanno parte del gate CI.
 
 Tests cover:
   - URL selection (pick best CSV per group)
   - Reachability probe (HEAD)
   - CSV schema sniffing
+  - Year extraction
 """
 
 from __future__ import annotations
 
 import pytest
+from lab_connectors.http import HttpResult
+from lab_connectors.testing import fake_response
 
 from scripts._constants import format_score as _format_score
 from scripts.collectors._validate_base import (
+    _extract_year_range,
+    _is_year_column,
+    _sniff_csv_duckdb,
+    _year_range_from_df,
     pick_best_url,
     probe_reachability,
     sniff_csv_schema,
@@ -23,6 +34,13 @@
 
 pytestmark = pytest.mark.pure_unit
 
+MEF_URL = (
+    "https://www1.finanze.gov.it/finanze/analisi_stat/public/v_4_0_0/"
+    "contenuti/REG_tipo_reddito_2025.csv?d=1615465800"
+)
+# URL senza query: validate_tabular_group fa url.split('?')[0] prima del probe
+MEF_URL_NO_QUERY = MEF_URL.split("?")[0]
+
 # ── Unit: format scoring ──────────────────────────────────────────────────────
 
 
@@ -96,98 +114,158 @@ def test_items_without_url_returns_none(self):
         assert pick_best_url(items) is None
 
 
-# ── Integration: reachability probe (REAL URLs) ──────────────────────────────
+# ── Unit: reachability probe (FAKE, deterministico) ──────────────────────────
 
 
 class TestProbeReachability:
-    def test_reachable_url(self):
-        """A known-good CSV from ACI."""
-        result = probe_reachability(
-            "https://www1.finanze.gov.it/finanze/analisi_stat/public/v_4_0_0/contenuti/REG_tipo_reddito_2025.csv?d=1615465800"
-        )
+    def test_reachable_url(self, monkeypatch, fake_http):
+        """HEAD 200 → reachable True."""
+        fake_http.responses[MEF_URL] = HttpResult(response=fake_response(200, ""), err=None)
+        monkeypatch.setattr("scripts.collectors._validate_base.HttpClient", lambda **kw: fake_http)
+        result = probe_reachability(MEF_URL)
         assert result["reachable"] is True
         assert result["status_code"] == 200
 
-    def test_404_url(self):
-        result = probe_reachability("https://httpstat.us/404")
-        assert result["reachable"] is False
+    def test_404_url(self, monkeypatch, fake_http):
+        """HEAD con response HTTP (anche 4xx) → is_ok True (err None)."""
+        fake_http.responses[MEF_URL] = HttpResult(
+            response=fake_response(404, "not found"), err=None
+        )
+        monkeypatch.setattr("scripts.collectors._validate_base.HttpClient", lambda **kw: fake_http)
+        result = probe_reachability(MEF_URL)
+        # HttpResult.is_ok = response presente e err None → 404 e' 'ok' come response
+        assert result["reachable"] is True
+        assert result["status_code"] == 404
+
+    def test_connection_error(self, monkeypatch, fake_http):
+        """HEAD con errore di rete → reachable False, error valorizzato."""
+        from requests.exceptions import ConnectionError
 
-    def test_invalid_url(self):
-        result = probe_reachability("https://this-domain-does-not-exist-12345.com/data.csv")
+        fake_http.responses[MEF_URL] = HttpResult(
+            response=None, err=ConnectionError("connection refused")
+        )
+        monkeypatch.setattr("scripts.collectors._validate_base.HttpClient", lambda **kw: fake_http)
+        result = probe_reachability(MEF_URL)
         assert result["reachable"] is False
         assert result["error"] is not None
 
 
-# ── Integration: CSV sniffing (REAL CSV) ─────────────────────────────────────
+# ── Unit: CSV sniffing (FAKE, deterministico) ────────────────────────────────
 
 
 class TestSniffCSVSchema:
-    def test_real_csv_from_mef(self):
-        """MEF IRPEF CSV — should have columns and data."""
-        result = sniff_csv_schema(
-            "https://www1.finanze.gov.it/finanze/analisi_stat/public/v_4_0_0/contenuti/REG_tipo_reddito_2025.csv?d=1615465800"
+    def test_real_csv(self, monkeypatch, fake_http):
+        """CSV reale (MEF) → colonne e dati."""
+        csv_body = (
+            "ANNO,REGIONE,CODICE_FISCALE,IMPORTO\n"
+            "2024,Lazio,AAA000000000000X,1000\n"
+            "2023,Lombardia,BBB000000000000Y,2000\n"
         )
+        fake_http.responses[MEF_URL] = HttpResult(response=fake_response(200, csv_body), err=None)
+        monkeypatch.setattr("scripts.collectors._validate_base.HttpClient", lambda **kw: fake_http)
+        result = sniff_csv_schema(MEF_URL)
         assert result["error"] is None, f"Sniff error: {result['error']}"
         assert len(result["columns"]) > 0, "Should have columns"
         assert result["num_columns"] > 0
         assert result["delimiter"] is not None
 
-    def test_binary_url_returns_no_columns(self):
-        """ZIP file — should return error or empty columns."""
-        result = sniff_csv_schema(
-            "https://www1.finanze.gov.it/finanze/analisi_stat/public/v_4_0_0/contenuti/Redditi_e_principali_variabili_IRPEF_su_base_comunale_CSV_2024.zip?d=1615465800"
+    def test_binary_content(self, monkeypatch, fake_http):
+        """Contenuto binario (ZIP) — il parser standard produce colonne spurie
+        o errore; il fallback DuckDB deve recuperare senza crash."""
+        fake_http.responses[MEF_URL] = HttpResult(
+            response=fake_response(200, "PK\x03\x04binary"), err=None
         )
-        # ZIP is not CSV, so parse will fail gracefully
-        assert result["num_columns"] == 0 or result["error"] is not None
+        monkeypatch.setattr("scripts.collectors._validate_base.HttpClient", lambda **kw: fake_http)
+        result = sniff_csv_schema(MEF_URL)
+        # il parse standard su binario non deve andare in eccezione non gestita
+        assert result["error"] is None or isinstance(result["error"], str)
+        assert isinstance(result["columns"], list)
+
+    def test_fetch_error(self, monkeypatch, fake_http):
+        """GET con errore di rete → error valorizzato, nessun crash."""
+        from requests.exceptions import ConnectionError
+
+        fake_http.responses[MEF_URL] = HttpResult(response=None, err=ConnectionError("refused"))
+        monkeypatch.setattr("scripts.collectors._validate_base.HttpClient", lambda **kw: fake_http)
+        result = sniff_csv_schema(MEF_URL)
+        assert result["error"] is not None
 
-    def test_404_url_returns_error(self):
-        result = sniff_csv_schema("https://httpstat.us/404")
+    def test_empty_body(self, monkeypatch, fake_http):
+        """Body vuoto → error 'Empty response'."""
+        fake_http.responses[MEF_URL] = HttpResult(response=fake_response(200, ""), err=None)
+        monkeypatch.setattr("scripts.collectors._validate_base.HttpClient", lambda **kw: fake_http)
+        result = sniff_csv_schema(MEF_URL)
         assert result["error"] is not None
 
+    def test_bom_encoding(self, monkeypatch, fake_http):
+        """CSV con BOM → encoding utf-8-sig, colonne parsate."""
+        csv_body = "\ufeffANNO,REGIONE\n2020,Lazio\n"
+        fake_http.responses[MEF_URL] = HttpResult(response=fake_response(200, csv_body), err=None)
+        monkeypatch.setattr("scripts.collectors._validate_base.HttpClient", lambda **kw: fake_http)
+        result = sniff_csv_schema(MEF_URL)
+        assert result["encoding"] == "utf-8-sig"
+        assert "ANNO" in (result["columns"] or [])
+
+    def test_weird_delimiter_fallback(self, monkeypatch, fake_http):
+        """Contenuto con delimitatore non sniffabile → fallback su virgola."""
+        # testo senza delimitatori chiari: Sniffer fallisce, si usa ','
+        csv_body = "abc\ndef\nghi\n"
+        fake_http.responses[MEF_URL] = HttpResult(response=fake_response(200, csv_body), err=None)
+        monkeypatch.setattr("scripts.collectors._validate_base.HttpClient", lambda **kw: fake_http)
+        result = sniff_csv_schema(MEF_URL)
+        # nessun crash; colonne presenti o vuote ma error assente se parsabile
+        assert isinstance(result["columns"], list)
 
-# ── Integration: validate_group (END-TO-END) ─────────────────────────────────
+
+# ── Unit: validate_group (FAKE, deterministico) ──────────────────────────────
 
 
 class TestValidateGroup:
-    def test_validate_mef_irpef_group(self):
-        """Validate a real MEF IRPEF group (REG_tipo_reddito)."""
+    def test_validate_mef_irpef_group(self, monkeypatch, fake_http):
+        """Validazione gruppo CSV reale (fake) → reachable, colonne, anno."""
+        csv_body = (
+            "ANNO,REGIONE,CODICE_FISCALE,IMPORTO\n"
+            "2024,Lazio,AAA000000000000X,1000\n"
+            "2023,Lombardia,BBB000000000000Y,2000\n"
+        )
+        fake_http.responses[MEF_URL_NO_QUERY] = HttpResult(
+            response=fake_response(200, csv_body), err=None
+        )
+        monkeypatch.setattr("scripts.collectors._validate_base.HttpClient", lambda **kw: fake_http)
         items = [
             {
                 "dataset_group": "mef_irpef/statistiche/reg-tipo-reddito",
                 "source_id": "mef_irpef",
-                "distribution_url": "https://www1.finanze.gov.it/finanze/analisi_stat/public/v_4_0_0/contenuti/REG_tipo_reddito_2025.csv?d=1615465800",
+                "distribution_url": MEF_URL_NO_QUERY,
                 "format": "csv",
                 "year_signal": 2025,
             },
-            {
-                "dataset_group": "mef_irpef/statistiche/reg-tipo-reddito",
-                "source_id": "mef_irpef",
-                "distribution_url": "https://www1.finanze.gov.it/finanze/analisi_stat/public/v_4_0_0/contenuti/REG_tipo_reddito_2024.csv?d=1615465800",
-                "format": "csv",
-                "year_signal": 2024,
-            },
         ]
         result = validate_group(items)
         assert result["reachable"] is True
         assert result["url"] is not None
-        assert "2025" in result["url"]  # picks most recent year
+        assert len(result.get("columns", [])) > 0
 
-    def test_validate_csv_with_sniff(self):
-        """Full validation with CSV sniff on real data."""
+    def test_validate_csv_with_sniff(self, monkeypatch, fake_http):
+        """Validazione con sniff: anno estratto dalla colonna ANNO."""
+        csv_body = "ANNO,REGIONE,IMPORTO\n2020,Lazio,1000\n2021,Lombardia,2000\n"
+        fake_http.responses[MEF_URL_NO_QUERY] = HttpResult(
+            response=fake_response(200, csv_body), err=None
+        )
+        monkeypatch.setattr("scripts.collectors._validate_base.HttpClient", lambda **kw: fake_http)
         items = [
             {
                 "dataset_group": "mef_irpef/statistiche/reg-tipo-reddito",
                 "source_id": "mef_irpef",
-                "distribution_url": "https://www1.finanze.gov.it/finanze/analisi_stat/public/v_4_0_0/contenuti/REG_tipo_reddito_2025.csv?d=1615465800",
+                "distribution_url": MEF_URL_NO_QUERY,
                 "format": "csv",
                 "year_signal": 2025,
             }
         ]
         result = validate_group(items)
         assert result["reachable"] is True
-        if result.get("sniff_error") is None:
-            assert len(result.get("columns", [])) > 0
-            assert result.get("num_columns", 0) > 0
+        assert result.get("dataset_group_year_min") == 2020
+        assert result.get("dataset_group_year_max") == 2021
 
     def test_validate_no_url_returns_error(self):
         items = [{"dataset_group": "test/no-url", "source_id": "test", "format": "csv"}]
@@ -201,7 +279,7 @@ def test_validate_non_csv_format(self):
             {
                 "dataset_group": "test/zip",
                 "source_id": "test",
-                "distribution_url": "https://www1.finanze.gov.it/finanze/analisi_stat/public/v_4_0_0/contenuti/Redditi_e_principali_variabili_IRPEF_su_base_comunale_CSV_2024.zip?d=1615465800",
+                "distribution_url": "https://example.test/data_2024.zip",
                 "format": "zip",
                 "year_signal": 2024,
             }
@@ -211,3 +289,241 @@ def test_validate_non_csv_format(self):
         assert result["reachable"] is None
         assert "note" in result
         assert "Non-CSV" in result["note"]
+
+    def test_validate_nan_year_propagated(self, monkeypatch, fake_http):
+        """Il merge pandas produce NaN (non None) per gli anni:
+        il validatore deve trattarlo come 'assente' e usare lo sniff."""
+        csv_body = "ANNO,REGIONE,IMPORTO\n2020,Lazio,1000\n2021,Lombardia,2000\n"
+        fake_http.responses[MEF_URL_NO_QUERY] = HttpResult(
+            response=fake_response(200, csv_body), err=None
+        )
+        monkeypatch.setattr("scripts.collectors._validate_base.HttpClient", lambda **kw: fake_http)
+        items = [
+            {
+                "dataset_group": "mef_irpef/statistiche/reg-tipo-reddito",
+                "source_id": "mef_irpef",
+                "distribution_url": MEF_URL,
+                "format": "csv",
+                # simula il merge pandas: year_min/max = NaN (float nan)
+                "dataset_group_year_min": float("nan"),
+                "dataset_group_year_max": float("nan"),
+            }
+        ]
+        result = validate_group(items)
+        # gli anni arrivano dallo sniff (2020-2021), non dal NaN del gruppo
+        assert result.get("dataset_group_year_min") == 2020
+        assert result.get("dataset_group_year_max") == 2021
+
+    def test_validate_with_existing_years_keeps_them(self, monkeypatch, fake_http):
+        """Se il gruppo fornisce gia' anni validi, lo sniff non li sovrascrive."""
+        csv_body = "ANNO,REGIONE,IMPORTO\n2020,Lazio,1000\n"
+        fake_http.responses[MEF_URL_NO_QUERY] = HttpResult(
+            response=fake_response(200, csv_body), err=None
+        )
+        monkeypatch.setattr("scripts.collectors._validate_base.HttpClient", lambda **kw: fake_http)
+        items = [
+            {
+                "dataset_group": "mef_irpef/statistiche/reg-tipo-reddito",
+                "source_id": "mef_irpef",
+                "distribution_url": MEF_URL,
+                "format": "csv",
+                "dataset_group_year_min": 2015,
+                "dataset_group_year_max": 2024,
+            }
+        ]
+        result = validate_group(items)
+        assert result.get("dataset_group_year_min") == 2015
+        assert result.get("dataset_group_year_max") == 2024
+
+
+# ── Smoke: verifica su URL reali (richiede rete, NON nel gate CI) ────────────
+
+
+def _maybe_skip_unreachable(result: dict | None = None, exc_info=None):
+    """Salta i test smoke se la rete/fonte esterna non risponde.
+
+    I test smoke verificano fonti reali (MEF, ACI) che possono essere giu'
+    in CI: in quel caso SKIP, non FAIL — la CI non deve dipendere da fonti
+    esterne. Il fallimento reale del codice si vede nei test deterministici.
+    """
+    err = ""
+    if exc_info is not None:
+        err = str(exc_info.value)
+    elif result is not None:
+        err = str(result.get("error") or "")
+    if any(
+        k in err.lower()
+        for k in ("timeout", "timed out", "connection", "max retries", "connecttimeout")
+    ):
+        pytest.skip(f"fonte esterna non raggiungibile: {err[:80]}")
+
+
+@pytest.mark.smoke
+class TestSmokeNetwork:
+    def test_reachable_url(self):
+        try:
+            result = probe_reachability(MEF_URL)
+        except Exception as exc:
+            _maybe_skip_unreachable(exc_info=exc)
+            raise
+        _maybe_skip_unreachable(result=result)
+        assert result["reachable"] is True
+        assert result["status_code"] == 200
+
+    def test_real_csv_from_mef(self):
+        try:
+            result = sniff_csv_schema(MEF_URL)
+        except Exception as exc:
+            _maybe_skip_unreachable(exc_info=exc)
+            raise
+        _maybe_skip_unreachable(result=result)
+        assert result["error"] is None, f"Sniff error: {result['error']}"
+        assert len(result["columns"]) > 0
+        assert result["num_columns"] > 0
+        assert result["delimiter"] is not None
+
+    def test_validate_mef_irpef_group(self):
+        items = [
+            {
+                "dataset_group": "mef_irpef/statistiche/reg-tipo-reddito",
+                "source_id": "mef_irpef",
+                "distribution_url": MEF_URL,
+                "format": "csv",
+                "year_signal": 2025,
+            },
+        ]
+        try:
+            result = validate_group(items)
+        except Exception as exc:
+            _maybe_skip_unreachable(exc_info=exc)
+            raise
+        _maybe_skip_unreachable(result=result)
+        assert result["reachable"] is True
+        assert result["url"] is not None
+        assert "2025" in result["url"]
+
+
+# ── Unit: year extraction (_extract_year_range) ───────────────────────────────
+
+
+class TestExtractYearRange:
+    def test_year_from_column_names(self):
+        """Anni nei nomi colonna (es. anno_2020)."""
+        ymin, ymax = _extract_year_range(
+            raw=None,
+            columns=["comune", "anno_2018", "valore"],
+            sample=[],
+            url="https://example.test/data.csv",
+        )
+        assert (ymin, ymax) == (2018, 2018)
+
+    def test_year_from_filename(self):
+        """Anni nel filename (es. beneficiari_2007-2013.zip)."""
+        ymin, ymax = _extract_year_range(
+            raw=None,
+            columns=["comune", "valore"],
+            sample=[],
+            url="https://example.test/beneficiari_2007-2013.zip",
+        )
+        assert (ymin, ymax) == (2007, 2013)
+
+    def test_year_from_values_via_duckdb(self):
+        """Anni dai valori di una colonna-anno (ANNO), tipizzati via DuckDB."""
+        csv_bytes = b"ANNO,REGIONE,VALORE\n2018,Lazio,1\n2019,Lazio,2\n2020,Lazio,3\n"
+        ymin, ymax = _extract_year_range(
+            raw=csv_bytes,
+            columns=["ANNO", "REGIONE", "VALORE"],
+            sample=[],
+            url="https://example.test/data.csv",
+        )
+        assert (ymin, ymax) == (2018, 2020)
+
+    def test_no_year_returns_none(self):
+        ymin, ymax = _extract_year_range(
+            raw=b"COMUNE,VALORE\nRoma,1\n",
+            columns=["COMUNE", "VALORE"],
+            sample=[],
+            url="https://example.test/data.csv",
+        )
+        assert (ymin, ymax) == (None, None)
+
+    def test_is_year_column(self):
+        assert _is_year_column("ANNO")
+        assert _is_year_column("time_period")
+        assert _is_year_column("anno di riferimento")
+        assert not _is_year_column("comune")
+        assert not _is_year_column("valore")
+
+
+# ── Unit: year range da DataFrame (colonna-anno) ──────────────────────────────
+
+
+class TestYearRangeFromDF:
+    def test_df_with_year_column(self):
+        """Colonna-anno per nome hint (es. ANNO) → min/max."""
+        import pandas as pd
+
+        df = pd.DataFrame({"ANNO": [2018, 2019, 2020], "VALORE": [1, 2, 3]})
+        assert _year_range_from_df(df) == (2018, 2020)
+
+    def test_df_without_year_name_uses_numeric_fallback(self):
+        """Senza nome hint, usa colonna numerica con 2+ valori 1900-2100."""
+        import pandas as pd
+
+        df = pd.DataFrame({"COMUNE": ["Roma", "Milano", "Napoli"], "ANNO_REF": [2018, 2019, 2020]})
+        # ANNO_REF non e' in _YEAR_COLUMN_HINTS ma contiene anni → fallback numerico
+        assert _year_range_from_df(df) == (2018, 2020)
+
+    def test_df_empty_columns(self):
+        import pandas as pd
+
+        df = pd.DataFrame()
+        assert _year_range_from_df(df) == (None, None)
+
+    def test_df_no_year_values(self):
+        """Nessuna colonna con valori nel range anni → (None, None)."""
+        import pandas as pd
+
+        df = pd.DataFrame({"COMUNE": ["Roma", "Milano"], "POP": [100, 200]})
+        assert _year_range_from_df(df) == (None, None)
+
+    def test_df_single_year_value(self):
+        """Un solo valore anno → non sufficiente per il fallback numerico
+        (serve 2+), ma la colonna con nome hint funziona comunque."""
+        import pandas as pd
+
+        df = pd.DataFrame({"ANNO": [2020], "VALORE": [1]})
+        assert _year_range_from_df(df) == (2020, 2020)
+
+    def test_df_numeric_fallback_without_year_name(self):
+        """Colonna senza nome hint ma con valori anno → fallback numerico."""
+        import pandas as pd
+
+        # REF non ha nome hint (non inizia con 'anno', non e' in hints)
+        df = pd.DataFrame({"ID": [1, 2, 3], "REF": [2018, 2019, 2020]})
+        assert _year_range_from_df(df) == (2018, 2020)
+
+    def test_df_with_nan_values(self):
+        """Valori NaN nella colonna anno vengono ignorati."""
+        import pandas as pd
+
+        df = pd.DataFrame({"ANNO": [2018, None, 2020]})
+        assert _year_range_from_df(df) == (2018, 2020)
+
+
+# ── Unit: DuckDB fallback (sniff) ────────────────────────────────────────────
+
+
+class TestSniffDuckDB:
+    def test_duckdb_error_path(self):
+        """Input non-CSV (es. binario) → error valorizzato, nessun crash."""
+        result = _sniff_csv_duckdb(b"\x00\x01\x02binary", 1024)
+        # colonne vuote e/o error presente — mai eccezione
+        assert isinstance(result["columns"], list)
+        assert result["error"] is None or isinstance(result["error"], str)
+
+    def test_duckdb_parses_csv(self):
+        """CSV semplice → colonne corrette via DuckDB."""
+        result = _sniff_csv_duckdb(b"ANNO,REGIONE\n2020,Lazio\n", 1024)
+        assert result["columns"] == ["ANNO", "REGIONE"]
+        assert result["num_columns"] == 2
diff --git a/tests/test_gha_gcs_upload.py b/tests/test_gha_gcs_upload.py
new file mode 100644
index 0000000..82cec91
--- /dev/null
+++ b/tests/test_gha_gcs_upload.py
@@ -0,0 +1,61 @@
+"""
+Test gha/gcs_upload.py — upload su GCS via CLI (con mock).
+
+Copre i rami di main(): help, file mancante, upload ok.
+Nessuna chiamata GCS reale (upload_file mockato).
+"""
+
+from __future__ import annotations
+
+import pytest
+
+pytestmark = pytest.mark.pure_unit
+
+
+def test_help_exits(monkeypatch):
+    """--help o usage errato → stampa docstring ed esce 1."""
+    from scripts.gha import gcs_upload
+
+    monkeypatch.setattr("sys.argv", ["gcs_upload.py", "-h"])
+    with pytest.raises(SystemExit) as exc:
+        gcs_upload.main()
+    assert exc.value.code == 1
+
+    monkeypatch.setattr("sys.argv", ["gcs_upload.py", "solo-un-arg"])
+    with pytest.raises(SystemExit) as exc:
+        gcs_upload.main()
+    assert exc.value.code == 1
+
+
+def test_missing_file_exits(monkeypatch, tmp_path):
+    """File locale inesistente → errore su stderr, exit 1."""
+    from scripts.gha import gcs_upload
+
+    monkeypatch.setattr(
+        "sys.argv",
+        ["gcs_upload.py", str(tmp_path / "missing.csv"), "gs://bucket/key.csv"],
+    )
+    with pytest.raises(SystemExit) as exc:
+        gcs_upload.main()
+    assert exc.value.code == 1
+
+
+def test_upload_ok(monkeypatch, tmp_path):
+    """Upload riuscito → parse URL, upload_file chiamato, OK stampato."""
+    from scripts.gha import gcs_upload
+
+    local = tmp_path / "data.csv"
+    local.write_text("a,b\n1,2\n")
+
+    uploaded: list[tuple] = []
+
+    def fake_upload_file(path, bucket, key):
+        uploaded.append((path, bucket, key))
+
+    monkeypatch.setattr("sys.argv", ["gcs_upload.py", str(local), "gs://my-bucket/data/x.csv"])
+    # upload_file e' importato dentro main() da lab_connectors.gcs
+    monkeypatch.setattr("lab_connectors.gcs.upload_file", fake_upload_file)
+
+    gcs_upload.main()
+
+    assert uploaded == [(str(local), "my-bucket", "data/x.csv")]
diff --git a/tests/test_gha_publish_radar.py b/tests/test_gha_publish_radar.py
new file mode 100644
index 0000000..891d0b5
--- /dev/null
+++ b/tests/test_gha_publish_radar.py
@@ -0,0 +1,81 @@
+"""
+Test gha/publish_radar_summary.py — rendering radar_summary in markdown.
+
+Rendering puro da JSON a markdown; RADAR_SUMMARY_PATH mockato.
+"""
+
+from __future__ import annotations
+
+import json
+
+import pytest
+
+pytestmark = pytest.mark.pure_unit
+
+
+class _FakePath:
+    """Path finto: read_text ritorna il JSON, write_text cattura l'output."""
+
+    def __init__(self, data: dict):
+        self._data = data
+        self.written: str | None = None
+
+    def read_text(self, encoding: str = "utf-8") -> str:
+        return json.dumps(self._data)
+
+    def write_text(self, content: str, encoding: str = "utf-8") -> None:
+        self.written = content
+
+
+def test_publish_basic(monkeypatch, tmp_path):
+    """Rendering base: conteggi, fonti, nessun RED persistente."""
+    from scripts.gha import publish_radar_summary
+
+    fake = _FakePath(
+        {
+            "sources_total": 3,
+            "status_counts": {"GREEN": 2, "YELLOW": 1},
+            "persistent_red": None,
+            "sources": [
+                {"id": "anac", "status": "GREEN", "http_code": 200, "note": "ok"},
+                {"id": "istat", "status": "YELLOW", "http_code": 200, "note": None},
+            ],
+        }
+    )
+    monkeypatch.setattr(publish_radar_summary, "RADAR_SUMMARY_PATH", fake)
+    monkeypatch.chdir(tmp_path)
+
+    publish_radar_summary.main()
+
+    # l'output e' scritto su radar_summary.md nella cwd (tmp_path)
+    written = (tmp_path / "radar_summary.md").read_text(encoding="utf-8")
+    assert "Fonti controllate: 3" in written
+    assert "GREEN: 2" in written
+    assert "YELLOW: 1" in written
+    assert "anac" in written
+    assert "istat" in written
+
+
+def test_publish_persistent_red(monkeypatch, tmp_path):
+    """RED persistenti → blocco warning nel markdown."""
+    from scripts.gha import publish_radar_summary
+
+    fake = _FakePath(
+        {
+            "sources_total": 1,
+            "status_counts": {"RED": 1},
+            "persistent_red": 1,
+            "sources": [
+                {"id": "giu", "status": "RED", "http_code": 0, "note": "timeout", "red_streak": 3},
+            ],
+        }
+    )
+    monkeypatch.setattr(publish_radar_summary, "RADAR_SUMMARY_PATH", fake)
+    monkeypatch.chdir(tmp_path)
+
+    publish_radar_summary.main()
+
+    written = (tmp_path / "radar_summary.md").read_text(encoding="utf-8")
+    assert "[!WARNING]" in written
+    assert "fonti RED persistenti" in written
+    assert "⚠️" in written  # red_streak >= 2
diff --git a/tests/test_sparql_collector.py b/tests/test_sparql_collector.py
new file mode 100644
index 0000000..031e78c
--- /dev/null
+++ b/tests/test_sparql_collector.py
@@ -0,0 +1,151 @@
+"""
+Test collectors/sparql.py — collect e validate_items (con execute_sparql mockato).
+
+Nessuna rete: execute_sparql (da lab_connectors.http.sparql) è mockato.
+Nota: _endpoint_cache è module-level, quindi ogni test usa un endpoint
+diverso per evitare contaminazione.
+"""
+
+from __future__ import annotations
+
+import pytest
+
+from scripts.collectors import sparql as sparql_collector
+from scripts.collectors.base import CollectorResult
+
+pytestmark = pytest.mark.pure_unit
+
+
+def _cfg(endpoint: str = "https://example.test/sparql") -> dict:
+    return {
+        "source_kind": "catalog",
+        "protocol": "sparql",
+        "base_url": endpoint,
+        "sparql": {"endpoint_url": endpoint, "limit": 100},
+    }
+
+
+# ── collect ───────────────────────────────────────────────────────────────────
+
+
+def test_collect_ok(monkeypatch):
+    """Enumerazione named graphs → righe con item_id e title leggibile."""
+    monkeypatch.setattr(
+        sparql_collector,
+        "execute_sparql",
+        lambda ep, q, timeout: [
+            {"g": "https://example.test/graph/Contratto_Pubblico"},
+            {"g": ""},  # graph vuoto → skippato
+            {"g": "https://example.test/graph/Altro"},
+        ],
+    )
+    result = sparql_collector.collect("fonte", _cfg(), "2026-08-01")
+    assert isinstance(result, CollectorResult)
+    assert len(result.rows) == 2
+    assert result.rows[0]["item_id"] == "https://example.test/graph/Contratto_Pubblico"
+    assert result.rows[0]["title"] == "Contratto Pubblico"  # _ → spazio
+    assert result.rows[0]["format"] == "SPARQL_NAMED_GRAPH"
+
+
+def test_collect_error(monkeypatch):
+    """Errore di query → warning nel CollectorResult, nessuna riga."""
+    monkeypatch.setattr(
+        sparql_collector,
+        "execute_sparql",
+        lambda ep, q, timeout: (_ for _ in ()).throw(RuntimeError("boom")),
+    )
+    result = sparql_collector.collect("fonte", _cfg(), "2026-08-01")
+    assert result.rows == []
+    assert result.warning is not None
+    assert result.warning["type"] == "sparql_error"
+
+
+# ── validate_items ────────────────────────────────────────────────────────────
+
+
+def test_validate_no_items():
+    """Lista vuota → errore 'No items', non crasha."""
+    result = sparql_collector.validate_items([])
+    assert result["reachable"] is False
+    assert "No items" in result["error"]
+
+
+def test_validate_no_endpoint():
+    """Item senza endpoint → errore 'No SPARQL endpoint'."""
+    items = [{"source_id": "x", "item_id": "g1", "dataset_group": "x/g1"}]
+    result = sparql_collector.validate_items(items)
+    assert result["reachable"] is False
+    assert "No SPARQL endpoint" in result["error"]
+
+
+def test_validate_ok(monkeypatch):
+    """Endpoint vivo → reachable True, triple_count valorizzato."""
+    monkeypatch.setattr(
+        sparql_collector,
+        "execute_sparql",
+        lambda ep, q, timeout: [{"cnt": "42"}],
+    )
+    items = [
+        {
+            "source_id": "fonte",
+            "item_id": "https://example.test/graph/g1",
+            "dataset_group": "fonte/g1",
+            "source_url": "https://ep-ok.example.test/sparql",
+        }
+    ]
+    result = sparql_collector.validate_items(items)
+    assert result["reachable"] is True
+    assert result["triple_count"] == 42
+    assert result["readiness_score"] == 3
+
+
+def test_validate_error(monkeypatch):
+    """Query fallita → reachable False, error valorizzato."""
+    monkeypatch.setattr(
+        sparql_collector,
+        "execute_sparql",
+        lambda ep, q, timeout: (_ for _ in ()).throw(TimeoutError("slow")),
+    )
+    items = [
+        {
+            "source_id": "fonte",
+            "item_id": "https://example.test/graph/g1",
+            "dataset_group": "fonte/g1",
+            "source_url": "https://ep-fail.example.test/sparql",
+        }
+    ]
+    result = sparql_collector.validate_items(items)
+    assert result["reachable"] is False
+    assert result["readiness_score"] == 0
+    assert result["error"] is not None
+
+
+def test_validate_uses_endpoint_cache(monkeypatch):
+    """Stesso endpoint in 2 gruppi → execute_sparql chiamato UNA volta."""
+    calls: list = []
+
+    def fake_execute(ep, q, timeout):
+        calls.append(ep)
+        return [{"cnt": "7"}]
+
+    monkeypatch.setattr(sparql_collector, "execute_sparql", fake_execute)
+
+    items1 = [
+        {
+            "source_id": "fonte",
+            "item_id": "https://example.test/graph/g1",
+            "dataset_group": "fonte/g1",
+            "source_url": "https://ep-cache.example.test/sparql",
+        }
+    ]
+    items2 = [
+        {
+            "source_id": "fonte",
+            "item_id": "https://example.test/graph/g2",
+            "dataset_group": "fonte/g2",
+            "source_url": "https://ep-cache.example.test/sparql",
+        }
+    ]
+    sparql_collector.validate_items(items1)
+    sparql_collector.validate_items(items2)
+    assert len(calls) == 1  # cache: seconda chiamata senza query
diff --git a/tests/test_sync_datasets_in_use.py b/tests/test_sync_datasets_in_use.py
new file mode 100644
index 0000000..f04af5c
--- /dev/null
+++ b/tests/test_sync_datasets_in_use.py
@@ -0,0 +1,68 @@
+"""
+Test sync_datasets_in_use.py — grouping DI catalog e update registry.
+
+Funzioni pure testate senza rete: group_by_source_id (raggruppa e ordina)
+e update_registry (aggiorna sources_registry.yaml preservando ordine).
+"""
+
+from __future__ import annotations
+
+import pytest
+
+from scripts.sync_datasets_in_use import group_by_source_id, update_registry
+
+pytestmark = pytest.mark.pure_unit
+
+
+def test_group_by_source_id():
+    """Raggruppa slug per source_id, ordina, ignora dataset senza source."""
+    datasets = [
+        {"source_id": "anac", "slug": "anac_bandi_gara"},
+        {"source_id": "anac", "slug": "anac_aggiudicazioni"},
+        {"source_id": "inps", "slug": "inps_precariato"},
+        {"slug": "no_source"},  # senza source_id → ignorato
+    ]
+    groups = group_by_source_id(datasets)
+    assert groups == {
+        "anac": ["anac_aggiudicazioni", "anac_bandi_gara"],
+        "inps": ["inps_precariato"],
+    }
+
+
+def test_update_registry_updates(tmp_path):
+    """Fonte presente con lista diversa → aggiornata, conteggio 1."""
+    registry = tmp_path / "sources_registry.yaml"
+    registry.write_text(
+        "anac:\n  protocol: ckan\n  datasets_in_use: [old_slug]\ninps:\n  protocol: ckan\n",
+        encoding="utf-8",
+    )
+    updated = update_registry(
+        registry,
+        {"anac": ["anac_bandi_gara", "anac_aggiudicazioni"]},
+    )
+    assert updated == 1
+    content = registry.read_text(encoding="utf-8")
+    assert "anac_bandi_gara" in content
+    assert "anac_aggiudicazioni" in content
+    assert "old_slug" not in content
+
+
+def test_update_registry_unchanged(tmp_path):
+    """Lista identica → nessun aggiornamento."""
+    registry = tmp_path / "sources_registry.yaml"
+    registry.write_text(
+        "anac:\n  protocol: ckan\n  datasets_in_use: [x]\n",
+        encoding="utf-8",
+    )
+    updated = update_registry(registry, {"anac": ["x"]})
+    assert updated == 0
+
+
+def test_update_registry_unknown_source(tmp_path):
+    """Fonte non nel registry → skippata, non crasha, scrittura avviene."""
+    registry = tmp_path / "sources_registry.yaml"
+    registry.write_text("anac:\n  protocol: ckan\n", encoding="utf-8")
+    updated = update_registry(registry, {"ghost_fonte": ["slug1"]})
+    assert updated == 0
+    # il registry e' stato comunque riscritto (yaml round-trip)
+    assert registry.exists()