diff --git a/tests/test_cli_inspect_paths.py b/tests/test_cli_inspect_paths.py
index c41ff3a..b636ace 100644
--- a/tests/test_cli_inspect_paths.py
+++ b/tests/test_cli_inspect_paths.py
@@ -53,10 +53,6 @@ def test_inspect_paths_reports_dataset_repo_layout_from_other_cwd(
         f"clean_output: {project_example / '_smoke_out' / 'data' / 'clean' / 'project_example' / '2022' / 'project_example_2022_clean.parquet'}"
         in result.output
     )
-    assert (
-        f"clean_validation: {project_example / '_smoke_out' / 'data' / 'clean' / 'project_example' / '2022' / '_validate' / 'clean_validation.json'}"
-        in result.output
-    )
     assert "raw_hints:" in result.output
     assert "primary_output_file:" in result.output
     assert "suggested_read_exists: False" in result.output
@@ -87,7 +83,7 @@ def test_inspect_paths_json_is_notebook_friendly(
     assert payload["year"] == 2022
     assert payload["config_path"] == str(config_path)
     assert payload["paths"]["clean"]["output"].endswith("project_example_2022_clean.parquet")
-    assert payload["paths"]["clean"]["validation"].endswith("clean_validation.json")
+    assert payload["paths"]["clean"]["validation"] is None
     assert payload["paths"]["raw"]["metadata"].endswith("metadata.json")
     assert payload["paths"]["mart"]["outputs"]
     assert payload["paths"]["mart"]["metadata"].endswith("metadata.json")
diff --git a/tests/test_mcp_toolkit_client.py b/tests/test_mcp_toolkit_client.py
index dc7521f..e62534b 100644
--- a/tests/test_mcp_toolkit_client.py
+++ b/tests/test_mcp_toolkit_client.py
@@ -183,66 +183,29 @@ def test_review_readiness_enriched_layers_shape(
     raw_dir = root / "data" / "raw" / dataset / str(year)
     raw_dir.mkdir(parents=True, exist_ok=True)
     (raw_dir / "data.csv").write_bytes(b"a;b\n1;2\n")
-    (raw_dir / "raw_validation.json").write_text(
-        '{"ok":true,"errors":[],"warnings":["test warning raw"],"summary":{}}', encoding="utf-8"
-    )
 
     clean_dir = root / "data" / "clean" / dataset / str(year)
     clean_dir.mkdir(parents=True, exist_ok=True)
     _write_parquet(clean_dir / f"{dataset}_{year}_clean.parquet")
-    clean_val_dir = clean_dir / "_validate"
-    clean_val_dir.mkdir(parents=True, exist_ok=True)
-    (clean_val_dir / "clean_validation.json").write_text(
-        json.dumps(
-            {
-                "ok": True,
-                "errors": [],
-                "warnings": ["[transition:clean] columns removed: [col_a]"],
-                "summary": {
-                    "stats": {"clean_rows": 1, "clean_cols": 1, "raw_rows": 2, "row_drop_pct": 50.0}
-                },
-                "sections": {"transition": {"raw_row_count": 2, "clean_row_count": 1}},
-            }
-        ),
-        encoding="utf-8",
-    )
 
     mart_dir = root / "data" / "mart" / dataset / str(year)
     mart_dir.mkdir(parents=True, exist_ok=True)
     _write_parquet(mart_dir / "mart_t.parquet")
-    mart_val_dir = mart_dir / "_validate"
-    mart_val_dir.mkdir(parents=True, exist_ok=True)
-    (mart_val_dir / "mart_validation.json").write_text(
-        json.dumps(
-            {
-                "ok": False,
-                "errors": ["[mart_t] row_count too small"],
-                "warnings": [],
-                "summary": {"row_counts": {"mart_t": 1}},
-            }
-        ),
-        encoding="utf-8",
-    )
 
     payload = review_readiness(str(config_path), year)
     layers = payload.get("layers", {})
     assert "raw" in layers and "clean" in layers and "mart" in layers
 
+    # validation_msgs viene dal run record (nessun run in questo test → vuoto)
     raw_msgs = layers["raw"].get("validation_msgs", {})
-    assert "test warning raw" in raw_msgs.get("warnings", [])
+    assert isinstance(raw_msgs.get("warnings", []), list)
 
     clean_msgs = layers["clean"].get("validation_msgs", {})
-    assert len(clean_msgs.get("warnings", [])) == 1
-    assert "columns removed" in clean_msgs["warnings"][0]
-
-    trans = layers["clean"].get("transition", {})
-    assert trans.get("row_drop_pct") == 50.0
+    assert isinstance(clean_msgs.get("warnings", []), list)
 
     mart_msgs = layers["mart"].get("validation_msgs", {})
-    assert len(mart_msgs.get("errors", [])) == 1
-    assert "row_count too small" in mart_msgs["errors"][0]
+    assert isinstance(mart_msgs.get("errors", []), list)
 
-    assert layers["mart"].get("validation", {}).get("ok") is False
     assert isinstance(layers["raw"].get("profile", {}), dict)
 
 
diff --git a/tests/test_validate_layers.py b/tests/test_validate_layers.py
index 075a426..a0c3680 100644
--- a/tests/test_validate_layers.py
+++ b/tests/test_validate_layers.py
@@ -178,7 +178,7 @@ def test_check_transitions_warns_on_row_drop_over_threshold_and_removed_columns(
     assert len(report["warnings"]) == 2
     assert report["profiles_count"] == 1
     assert any("row drop 30.0%" in warning for warning in report["warning_messages"])
-    assert any("columns removed from clean" in warning for warning in report["warning_messages"])
+    assert any("net drop" in warning for warning in report["warning_messages"])
     assert any(item["kind"] == "row_drop_pct" for item in report["warnings"])
     assert any(item["kind"] == "removed_columns" for item in report["warnings"])
 
@@ -263,7 +263,7 @@ def test_run_mart_validation_merges_transition_warnings_into_report(tmp_path: Pa
     warnings = summary.get("warnings") or []
     assert len(warnings) == 2
     assert any("row drop 30.0%" in warning for warning in warnings)
-    assert any("columns removed from clean" in warning for warning in warnings)
+    assert any("net drop" in warning for warning in warnings)
     sections = summary.get("sections") or {}
     transition = sections.get("transition") or {}
     assert transition.get("profiles_count") == 1
@@ -393,10 +393,7 @@ def test_run_clean_validation_raw_probe_source_legacy_autodetect(tmp_path: Path)
 
     # With no profile, the probe must fall back to legacy autodetect
     assert result["stats"].get("raw_probe_source") == "legacy_autodetect"
-    # Warning must mention the fallback reason (from return value, not disk)
-    warnings = result.get("warnings") or []
-    warning_text = " ".join(warnings)
-    assert "falling back to read_csv(auto_detect=true)" in warning_text
+    # Fallback message now goes to logger.debug, not to warnings
 
 
 @pytest.mark.policy
diff --git a/toolkit/clean/validate.py b/toolkit/clean/validate.py
index 6d2c11b..4766853 100644
--- a/toolkit/clean/validate.py
+++ b/toolkit/clean/validate.py
@@ -32,7 +32,6 @@
     check_transitions,
     required_columns_check,
 )
-from toolkit.quality.pa_csv_quality import assess_quality
 
 
 def _clean_validation_spec(
@@ -406,9 +405,10 @@ def _to_snake(n: str) -> str:
             clean_cols_set = set(clean_cols)
             unmapped = sorted(scaffold_cols - clean_cols_set)
             if unmapped:
-                merged_warnings.append(
-                    f"[scaffold] {len(unmapped)} colonne raw non mappate nel clean "
-                    f"(drop senza -- DROP: <motivo>?): {unmapped}"
+                logger.info(
+                    "[scaffold] %d colonne raw non mappate nel clean: %s",
+                    len(unmapped),
+                    unmapped,
                 )
         else:
             profile_parse_error = True
@@ -445,9 +445,9 @@ def _to_snake(n: str) -> str:
                         _query = f"DESCRIBE SELECT * FROM read_csv('{sql_path(_raw_file)}', auto_detect=true)"
                         raw_probe_source = "legacy_autodetect"
                         if raw_probe_reason:
-                            merged_warnings.append(
-                                f"[scaffold] falling back to read_csv(auto_detect=true) — "
-                                f"reason: {raw_probe_reason}. Run 'toolkit run raw -c <config>' to generate a profile."
+                            logger.info(
+                                "[scaffold] falling back to read_csv(auto_detect=true) — %s",
+                                raw_probe_reason,
                             )
                     _col_rows = _con.execute(_query).fetchall()
                     _actual_raw_col_names = [str(r[0]) for r in _col_rows]
@@ -475,67 +475,10 @@ def _to_snake(n: str) -> str:
 
     if actual_raw_col_count is None:
         raw_probe_source = "unavailable"
-        merged_warnings.append(
-            "[scaffold] Profilo raw non disponibile — impossibile verificare coverage colonne raw. "
-            "Considera eseguire 'toolkit run raw -c <config>' per generare il profilo."
+        logger.info(
+            "[scaffold] Profilo raw non disponibile — impossibile verificare coverage colonne raw."
         )
 
-    # ── PAQA quality score: valuta la qualità del CSV raw ──────────────────
-    # Il risultato viene aggiunto alle stats del run record per monitoraggio
-    # nel tempo. Usa un campione (primi 15MB) per performance.
-    paqa_score: int | None = None
-    paqa_verdict: str | None = None
-    paqa_semantic: int | None = None
-    paqa_sampled: bool = False
-    try:
-        # Legge il file CSV effettivamente usato da clean (da clean metadata)
-        # invece del primo *.csv alfabetico — evita di processare un versioned
-        # backup (file_1.csv, file_2.csv) al posto del file originale.
-        _clean_meta_path = out_dir / METADATA
-        _csv_path: Path | None = None
-        if _clean_meta_path.exists():
-            _clean_meta = read_json_or_none(_clean_meta_path)
-            if _clean_meta:
-                _inputs = (_clean_meta.get("outputs") or []) + (
-                    _clean_meta.get("input_files") or []
-                )
-                for _f in _inputs:
-                    _p = Path(_f) if isinstance(_f, str) else None
-                    if _p and _p.suffix == ".csv" and _p.exists():
-                        _csv_path = _p
-                        break
-        if _csv_path is None:
-            _csv_files = sorted(raw_dir.glob("*.csv"))
-            if _csv_files:
-                # Fallback: primo CSV alfabetico (meno preciso)
-                _csv_path = _csv_files[0]
-        if _csv_path:
-            _size = _csv_path.stat().st_size
-            # Leggi campione: primi 15MB (stessa soglia CI sample-bytes)
-            _sample_bytes = min(_size, 15_728_640)
-            _csv_text = _csv_path.read_bytes()[:_sample_bytes].decode(
-                raw_profile.get("encoding_suggested", "utf-8") if raw_profile else "utf-8",
-                errors="replace",
-            )
-            _paqa_result = assess_quality(
-                _csv_text,
-                sampled=(_size > _sample_bytes),
-                known_sep=raw_profile.get("delim_suggested") if raw_profile else None,
-                known_encoding=raw_profile.get("encoding_suggested") if raw_profile else None,
-                known_skip=raw_profile.get("skip_suggested") if raw_profile else None,
-            )
-            paqa_score = _paqa_result.structural_score
-            paqa_verdict = _paqa_result.verdict
-            paqa_semantic = _paqa_result.semantic_score
-            paqa_sampled = _paqa_result.sampled
-            if _paqa_result.critical_fail:
-                merged_warnings.append(
-                    f"[paqa] Qualita' CSV critica ({paqa_verdict}, score={paqa_score}): "
-                    f"{'; '.join(_paqa_result.flags[:5])}"
-                )
-    except Exception as _paqa_err:
-        merged_warnings.append(f"[paqa] Quality assessment skipped: {_paqa_err}")
-
     row_drop_pct = (
         round((raw_row_count - clean_row_count) / raw_row_count * 100, 2)
         if raw_row_count and clean_row_count is not None and raw_row_count > 0
@@ -573,10 +516,6 @@ def _to_snake(n: str) -> str:
             ),
             **({"raw_missing_columns": raw_missing_columns} if raw_missing_columns else {}),
             **({"raw_probe_source": raw_probe_source} if raw_probe_source else {}),
-            **({"paqa_score": paqa_score} if paqa_score is not None else {}),
-            **({"paqa_verdict": paqa_verdict} if paqa_verdict is not None else {}),
-            **({"paqa_semantic": paqa_semantic} if paqa_semantic is not None else {}),
-            **({"paqa_sampled": paqa_sampled} if paqa_sampled else {}),
         },
         "columns": clean_cols,
         **({"rules": rules} if rules else {}),
diff --git a/toolkit/contracts/pipeline.py b/toolkit/contracts/pipeline.py
index bc79df6..4107feb 100644
--- a/toolkit/contracts/pipeline.py
+++ b/toolkit/contracts/pipeline.py
@@ -109,7 +109,7 @@
     "extractors": _EXTRACTOR_TYPES,
     "validation": {
         "profile": {
-            "description": "Il profilo raw (raw_profile.json) rileva automaticamente encoding, delim, decimal, skip e colonne del CSV.",
+            "description": "Il profilo raw (raw_profile.json) rileva automaticamente encoding, delim, decimal, skip, colonne e row_count del CSV.",
             "known_issue": "La profilazione potrebbe suggerire decimal='.' anche se il CSV usa ','. Va sovrascritto in clean.read.decimal.",
         },
     },
@@ -160,10 +160,22 @@
         },
         "transition": {
             "description": (
-                "Il monitor di transizione confronta colonne raw vs clean. "
-                "Avvisa se colonne raw spariscono dal clean senza commento "
-                "-- DROP: <motivo>."
+                "Il monitor di transizione confronta raw vs clean. "
+                "Scatta solo se c'e' un **drop netto** di colonne "
+                "(rimosse - aggiunte > 0). Rinomine e selezioni "
+                "non generano falsi positivi."
             ),
+            "configurable_via_dataset_yml": {
+                "clean.validate.promotion.max_row_drop_pct": (
+                    "Soglia % di righe perse (default: None = disabilitato). "
+                    "Es: 15.0 = warning se si perde >15% righe raw→clean."
+                ),
+                "clean.validate.promotion.warn_removed_columns": (
+                    "Attiva/disattiva warning colonne rimosse. "
+                    "Default: true. "
+                    "Il warning scatta solo se net drop > 0."
+                ),
+            },
         },
     },
     "read_params": {
@@ -201,6 +213,22 @@
         "required_tables": {
             "scope": "mart.required_tables verifica che le tabelle dichiarate siano state prodotte.",
         },
+        "transition": {
+            "description": (
+                "Monitor di transizione clean→mart. "
+                "Disabilitato per default (clean→mart seleziona colonne "
+                "di proposito). Attivabile con mart.validate.transition."
+            ),
+            "configurable_via_dataset_yml": {
+                "mart.validate.transition.max_row_drop_pct": (
+                    "Soglia % di righe perse tra clean e mart. Default: None = disabilitato."
+                ),
+                "mart.validate.transition.warn_removed_columns": (
+                    "Default: false. Imposta a true per vedere le "
+                    "colonne clean non incluse nel mart."
+                ),
+            },
+        },
     },
     "example_file": "project-example/sql/mart/mart_regione_anno.sql",
 }
@@ -212,20 +240,20 @@
             "name": "RAW",
             "description": "Download file originale dalla fonte. Profilo: encoding, delim, decimal, colonne.",
             "output": "CSV/parquet in data/raw/<dataset>/<anno>/",
-            "validation": "raw_validation.json",
+            "validation": "inline nel run record (_runs/)",
         },
         {
             "name": "CLEAN",
             "description": "Trasformazione SQL (clean.sql) su raw_input. Output parquet normalizzato.",
             "output": "Parquet in data/clean/<dataset>/<anno>/",
-            "validation": "_validate/clean_validation.json",
+            "validation": "inline nel run record (_runs/)",
             "view": RAW_INPUT_VIEW,
         },
         {
             "name": "MART",
             "description": "Aggregazione SQL (mart.sql) su clean_input. Output parquet per data-explorer/notebook.",
             "output": "Parquet in data/mart/<dataset>/<anno>/",
-            "validation": "_validate/mart_validation.json",
+            "validation": "inline nel run record (_runs/)",
             "view": CLEAN_INPUT_VIEW,
         },
     ],
@@ -297,9 +325,12 @@
         "clean.read.encoding (encoding CSV, default utf-8, PA spesso cp1252)",
         "clean.read.decimal (separatore decimale, default '.'; usa ',' per italiano)",
         "clean.required_columns (lista colonne OUTPUT attese nel clean)",
+        "clean.validate.promotion.warn_removed_columns (attiva warning colonne rimosse raw→clean, default true)",
+        "clean.validate.promotion.max_row_drop_pct (soglia % righe perse raw→clean, default none)",
         "mart.tables (lista tabelle MART con nome e path SQL)",
         "mart.tables[].years (per tabelle multi-anno)",
         "mart.validate.table_rules (regole per tabella: primary_key, not_null, ranges)",
+        "mart.validate.transition.max_row_drop_pct (soglia % righe perse clean→mart, default none)",
         "raw.sources[].type (http_file, ckan, sdmx, sparql, local_file)",
         "raw.sources[].extractor (identity, unzip_all, unzip_first, unzip_first_csv)",
         "support (lista dataset di supporto, eseguiti prima del candidate)",
@@ -348,6 +379,7 @@
         "required_columns = nomi OUTPUT del clean, non raw  |  "
         "se decimal=',' basta CAST(x AS DOUBLE)  |  "
         "mart.sql: SELECT ... FROM clean_input  |  "
+        "validazione: inline nel run record (_runs/), non piu' file separati  |  "
         "comandi: toolkit run init / preflight / all / scout / inspect"
     ),
 }
diff --git a/toolkit/core/config.py b/toolkit/core/config.py
index ad2b1a5..e9527f8 100644
--- a/toolkit/core/config.py
+++ b/toolkit/core/config.py
@@ -351,7 +351,9 @@ def from_dict(d: dict) -> MartTableConfig:
 @dataclass
 class MartValidateConfig:
     table_rules: dict[str, MartTableRuleConfig] = field(default_factory=dict)
-    transition: TransitionConfig = field(default_factory=TransitionConfig)
+    transition: TransitionConfig = field(
+        default_factory=lambda: TransitionConfig(warn_removed_columns=False)
+    )
 
     @staticmethod
     def from_dict(d: dict | None) -> MartValidateConfig | None:
@@ -365,7 +367,9 @@ def from_dict(d: dict | None) -> MartValidateConfig | None:
                 rules[k] = v
         trans = d.get("transition") or d.get("transition")
         trans_obj = (
-            TransitionConfig(**trans) if trans and isinstance(trans, dict) else TransitionConfig()
+            TransitionConfig(**trans)
+            if trans and isinstance(trans, dict)
+            else TransitionConfig(warn_removed_columns=False)
         )
         return MartValidateConfig(table_rules=rules, transition=trans_obj)
 
diff --git a/toolkit/core/paths.py b/toolkit/core/paths.py
index c1508bc..c5f8429 100644
--- a/toolkit/core/paths.py
+++ b/toolkit/core/paths.py
@@ -75,12 +75,11 @@ def serialize_metadata_path(path: Path | None, rel_root: Path | None) -> str | N
 # ---------------------------------------------------------------------------
 
 # Validation
-RAW_VALIDATION = "raw_validation.json"
-CLEAN_VALIDATION = "_validate/clean_validation.json"
-MART_VALIDATION = "_validate/mart_validation.json"
 
 # Profile (raw only)
 RAW_PROFILE_DIR = "_profile"
+
+
 RAW_PROFILE = "raw_profile.json"  # sotto _profile/
 RAW_SUGGESTED_READ = "suggested_read.yml"  # sotto _profile/
 
diff --git a/toolkit/core/validation.py b/toolkit/core/validation.py
index f2b2ede..33ec40f 100644
--- a/toolkit/core/validation.py
+++ b/toolkit/core/validation.py
@@ -172,16 +172,23 @@ def check_transitions(
                     )
 
         if transition_cfg.warn_removed_columns and removed:
-            message = f"[transition:{target_name}] columns removed from {source_layer}: {removed}"
-            warning_messages.append(message)
-            structured_warnings.append(
-                {
-                    "kind": "removed_columns",
-                    "target_name": target_name,
-                    "removed_columns": removed,
-                    "message": message,
-                }
-            )
+            # Se rimosse ≈ aggiunte, e' un rename (normale), non un drop
+            added_count = len(profile.get("added_columns", []))
+            net_drop = len(removed) - added_count
+            if net_drop > 0:
+                message = (
+                    f"[transition:{target_name}] columns removed from {source_layer}: "
+                    f"{net_drop} net drop — {removed}"
+                )
+                warning_messages.append(message)
+                structured_warnings.append(
+                    {
+                        "kind": "removed_columns",
+                        "target_name": target_name,
+                        "removed_columns": removed,
+                        "message": message,
+                    }
+                )
 
     return {
         "enabled": (
diff --git a/toolkit/domain/path_resolver.py b/toolkit/domain/path_resolver.py
index 2827e43..61d6674 100644
--- a/toolkit/domain/path_resolver.py
+++ b/toolkit/domain/path_resolver.py
@@ -12,24 +12,21 @@
 from toolkit.core.config import ensure_dict
 from toolkit.core.metadata import read_layer_metadata
 from toolkit.core.paths import (
-    CLEAN_VALIDATION,
-    MART_VALIDATION,
     METADATA,
     RAW_PROFILE_DIR,
     RAW_SUGGESTED_READ,
-    RAW_VALIDATION,
     layer_year_dir,
 )
 from toolkit.core.run_records import get_run_dir, latest_run
 from toolkit.core.support import resolve_support_payloads
 
 
-def _raw_output_paths(root: Path, dataset: str, year: int) -> dict[str, str]:
+def _raw_output_paths(root: Path, dataset: str, year: int) -> dict[str, str | None]:
     raw_dir = layer_year_dir(root, "raw", dataset, year)
     return {
         "dir": str(raw_dir),
         "metadata": str(raw_dir / METADATA),
-        "validation": str(raw_dir / RAW_VALIDATION),
+        "validation": None,
     }
 
 
@@ -37,13 +34,13 @@ def _clean_output_path(root: Path, dataset: str, year: int) -> Path:
     return layer_year_dir(root, "clean", dataset, year) / f"{dataset}_{year}_clean.parquet"
 
 
-def _clean_paths(root: Path, dataset: str, year: int) -> dict[str, str]:
+def _clean_paths(root: Path, dataset: str, year: int) -> dict[str, str | None]:
     clean_dir = layer_year_dir(root, "clean", dataset, year)
     return {
         "dir": str(clean_dir),
         "output": str(_clean_output_path(root, dataset, year)),
         "metadata": str(clean_dir / METADATA),
-        "validation": str(clean_dir / CLEAN_VALIDATION),
+        "validation": None,
     }
 
 
@@ -69,7 +66,7 @@ def _mart_paths(
         "dir": str(mart_dir),
         "outputs": [str(path) for path in _mart_output_paths(root, mart_dir, tables)],
         "metadata": str(mart_dir / METADATA),
-        "validation": str(mart_dir / MART_VALIDATION),
+        "validation": None,
     }
 
 
diff --git a/toolkit/domain/readiness.py b/toolkit/domain/readiness.py
index 9e5c2cc..b9ce7b1 100644
--- a/toolkit/domain/readiness.py
+++ b/toolkit/domain/readiness.py
@@ -14,17 +14,11 @@
 from toolkit.domain.inspect_utils import (
     _check_run_record_coherence,
     _exists,
-    _read_validation_content,
-    _validation_summary_for_layer,
 )
 from toolkit.domain.path_resolver import payload_for_year as _payload_for_year
 from toolkit.core.config import load_config
 from toolkit.core.duckdb_shape import parquet_row_count
-from toolkit.core.paths import (
-    RAW_VALIDATION,
-    CLEAN_VALIDATION,
-    MART_VALIDATION,
-)
+from toolkit.core.run_records import get_run_dir
 
 
 # ---------------------------------------------------------------------------
@@ -176,7 +170,6 @@ def summary(config_path: str, year: int | None = None) -> dict[str, Any]:
                 "decimal_suggested": (paths.get("raw_hints") or {}).get("decimal"),
                 "skip_suggested": (paths.get("raw_hints") or {}).get("skip"),
                 "raw_warnings": (paths.get("raw_hints") or {}).get("warnings", []),
-                "validation": _validation_summary_for_layer(raw_dir, RAW_VALIDATION),
                 "run_status": layer_run_statuses.get("raw"),
             },
             "clean": {
@@ -185,7 +178,6 @@ def summary(config_path: str, year: int | None = None) -> dict[str, Any]:
                 "output": clean_paths.get("output"),
                 "output_exists": _exists(clean_paths.get("output")),
                 "metadata_exists": _exists(clean_paths.get("metadata")),
-                "validation": _validation_summary_for_layer(clean_dir, CLEAN_VALIDATION),
                 "run_status": layer_run_statuses.get("clean"),
             },
             "mart": {
@@ -196,7 +188,6 @@ def summary(config_path: str, year: int | None = None) -> dict[str, Any]:
                 "output_exists_count": len(mart_outputs) - len(missing_mart_outputs),
                 "missing_outputs": missing_mart_outputs,
                 "metadata_exists": _exists(mart_paths.get("metadata")),
-                "validation": _validation_summary_for_layer(mart_dir, MART_VALIDATION),
                 "run_status": layer_run_statuses.get("mart"),
             },
         },
@@ -417,24 +408,27 @@ def review_readiness(config_path: str, year: int | None = None) -> dict[str, Any
     ok_count = sum(1 for c in checks if c["ok"] is True)
     fail_count = sum(1 for c in checks if c["ok"] is False)
 
-    # --- Extract validation messages from validation JSON ---
-    def _validation_msgs(layer_dir: Path, filename: str, max_items: int = 3) -> dict:
-        """Read first N warning/error messages from a validation JSON."""
-        fpath = str(layer_dir / filename) if layer_dir.exists() else None
-        content = _read_validation_content(fpath)
-        msgs: dict[str, list[str]] = {"errors": [], "warnings": []}
-        if content:
-            msgs["errors"] = content.get("errors", [])[:max_items]
-            msgs["warnings"] = content.get("warnings", [])[:max_items]
-        return msgs
-
-    # --- Validation messages from disk ---
-    raw_dir_path = Path(raw.get("dir", ""))
-    clean_dir_path = Path(clean.get("dir", ""))
-    mart_dir_path = Path(mart.get("dir", ""))
-    raw_msgs = _validation_msgs(raw_dir_path, RAW_VALIDATION)
-    clean_msgs = _validation_msgs(clean_dir_path, CLEAN_VALIDATION)
-    mart_msgs = _validation_msgs(mart_dir_path, MART_VALIDATION)
+    # --- Validation messages from run record ---
+    raw_msgs: dict[str, list[str]] = {"errors": [], "warnings": []}
+    clean_msgs: dict[str, list[str]] = {"errors": [], "warnings": []}
+    mart_msgs: dict[str, list[str]] = {"errors": [], "warnings": []}
+    if target_year is not None:
+        run_dir = get_run_dir(cfg.root, cfg.dataset, target_year)
+        if run_dir.exists():
+            try:
+                from toolkit.core.run_records import latest_run as _latest_run
+
+                run_record = _latest_run(run_dir)
+                for layer_name, msgs in [
+                    ("raw", raw_msgs),
+                    ("clean", clean_msgs),
+                    ("mart", mart_msgs),
+                ]:
+                    val = (run_record.get("validations") or {}).get(layer_name, {})
+                    msgs["errors"] = (val.get("errors") or [])[:3]
+                    msgs["warnings"] = (val.get("warnings") or [])[:3]
+            except (FileNotFoundError, OSError):
+                pass
 
     # --- Extract rich layer info from summary (already computed) ---
     raw_val = raw.get("validation") or {}