Fase 1: report unificato — validazione in memoria, non più file su disco

.gitignore

@@ -38,3 +38,5 @@ tmp/
 generated/
 /sql/
 *.py,cover
+data/
+datasets/

tests/helpers_assert_paths.py

@@ -160,9 +160,9 @@ def assert_golden_path_artifacts(
     """Verifica la struttura completa degli artifact dopo un golden path.
 
     Controlla che per ogni anno esistano:
-    - RAW dir con raw_validation.json
+    - RAW dir con metadata.json
     - CLEAN dir con almeno un parquet e metadata.json
-    - MART dir con ogni tabella dichiarata e mart_validation.json
+    - MART dir con ogni tabella dichiarata
 
     Args:
         mart_tables: Lista di nomi tabelle mart (senza estensione).
@@ -171,10 +171,7 @@ def assert_golden_path_artifacts(
 
     for year in years:
         # RAW
-        raw_dir = assert_raw_dir(root, dataset, year)
-        assert (raw_dir / "raw_validation.json").exists(), (
-            f"raw_validation.json mancante in {raw_dir}"
-        )
+        assert_raw_dir(root, dataset, year)
 
         # CLEAN
         clean_dir = assert_clean_dir(root, dataset, year)
@@ -189,6 +186,3 @@ def assert_golden_path_artifacts(
                 assert (mart_dir / f"{table}.parquet").exists(), (
                     f"MART table {table}.parquet mancante in {mart_dir}"
                 )
-            assert (mart_dir / "_validate" / "mart_validation.json").exists(), (
-                f"mart_validation.json mancante in {mart_dir}"
-            )

tests/test_pipeline_integration.py

@@ -80,8 +80,6 @@ def test_artifacts_policy_minimal_behaves_like_standard(project_example: Path, m
     assert (raw_dir / "metadata.json").exists()
     assert (clean_dir / "metadata.json").exists()
     assert (mart_dir / "metadata.json").exists()
-    assert (mart_dir / "metadata.json").exists()
-    assert (mart_dir / "_validate" / "mart_validation.json").exists()
 
 
 def test_artifacts_policy_standard_keeps_expected_artifacts(

tests/test_validate_layers.py

@@ -260,21 +260,22 @@ def test_run_mart_validation_merges_transition_warnings_into_report(tmp_path: Pa
     assert summary["passed"] is True
     assert summary["warnings_count"] == 2
 
-    report = json.loads(
-        (mart_dir / "_validate" / "mart_validation.json").read_text(encoding="utf-8")
-    )
-    assert len(report["warnings"]) == 2
-    assert any("row drop 30.0%" in warning for warning in report["warnings"])
-    assert any("columns removed from clean" in warning for warning in report["warnings"])
-    assert report["transition"]["profiles_count"] == 1
-    assert report["transition"]["warnings_count"] == 2
-    assert report["transition"]["config"] == {
+    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)
+    sections = summary.get("sections") or {}
+    transition = sections.get("transition") or {}
+    assert transition.get("profiles_count") == 1
+    assert transition.get("warnings_count") == 2
+    assert transition.get("config") == {
         "max_row_drop_pct": 20.0,
         "warn_removed_columns": True,
         "fail_on_row_drop_exceeded": False,
     }
-    assert any(item["kind"] == "row_drop_pct" for item in report["transition"]["warnings"])
-    assert any(item["kind"] == "removed_columns" for item in report["transition"]["warnings"])
+    transition_warnings = transition.get("warnings") or []
+    assert any(item["kind"] == "row_drop_pct" for item in transition_warnings)
+    assert any(item["kind"] == "removed_columns" for item in transition_warnings)
 
 
 @pytest.mark.policy
@@ -392,12 +393,10 @@ 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
-    # build_validation_summary only includes the raw stats, not the full result.warnings
-    # So we check the warning was emitted by inspecting via the ValidationResult if accessible,
-    # or by verifying that the fallback path was taken (raw_probe_source = legacy_autodetect)
-    warning_texts = " ".join(_read_warnings_from_validation_report(clean_dir))
-    assert "falling back to read_csv(auto_detect=true)" in warning_texts
+    # 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
 
 
 @pytest.mark.policy
@@ -445,7 +444,7 @@ def test_run_clean_validation_raw_probe_source_unavailable_when_no_raw_file(
 
 
 def _read_warnings_from_validation_report(clean_dir: Path) -> list[str]:
-    """Read warnings from the written validation JSON."""
+    """Read warnings from the written validation JSON (legacy, file may not exist)."""
     report_path = clean_dir / "_validate" / "clean_validation.json"
     if report_path.exists():
         data = json.loads(report_path.read_text(encoding="utf-8"))
@@ -687,12 +686,10 @@ def test_run_mart_validation_transition_errors_set_ok_false(tmp_path: Path):
     )
     assert summary["errors_count"] >= 1, "Expected at least 1 transition error"
 
-    # Verifica che il JSON di validazione su disco contenga l'errore
-    report = json.loads(
-        (mart_dir / "_validate" / "mart_validation.json").read_text(encoding="utf-8")
-    )
-    assert len(report.get("errors", [])) >= 1
-    assert any("row drop" in e for e in report["errors"])
+    # Verifica che il summary contenga l'errore
+    errors = summary.get("errors") or []
+    assert len(errors) >= 1
+    assert any("row drop" in e for e in errors)
 
 
 # ---------------------------------------------------------------------------

toolkit/clean/run.py

@@ -16,7 +16,6 @@
     write_metadata,
 )
 from toolkit.core.paths import (
-    CLEAN_VALIDATION,
     layer_year_dir,
     resolve_root,
     resolve_sql_path,
@@ -296,7 +295,6 @@ def run_clean(
     merge_layer_manifest(
         out_dir,
         metadata_path=metadata_path.name,
-        validation_path=CLEAN_VALIDATION,
         outputs=outputs,
     )
     logger.info(f"CLEAN -> {output_path}")

toolkit/clean/validate.py

@@ -20,9 +20,7 @@
 from toolkit.clean._helpers import _input_files_from_clean_metadata, _profile_raw_input
 from toolkit.core.config import CleanValidationSpec, RangeRuleConfig, TransitionConfig
 from toolkit.core.layer_profile import compare_layer_profiles
-from toolkit.core.metadata import merge_layer_manifest
 from toolkit.core.paths import (
-    CLEAN_VALIDATION,
     METADATA,
     RAW_PROFILE,
     layer_year_dir,
@@ -33,7 +31,6 @@
     build_validation_summary,
     check_transitions,
     required_columns_check,
-    write_validation_json,
 )
 from toolkit.quality.pa_csv_quality import assess_quality
 
@@ -593,16 +590,5 @@ def _to_snake(n: str) -> str:
         sections=merged_sections,
     )
 
-    report = write_validation_json(Path(out_dir) / CLEAN_VALIDATION, result)
-    metadata = read_json_or_none(out_dir / METADATA) or {}
-    merge_layer_manifest(
-        out_dir,
-        metadata_path=METADATA,
-        validation_path="_validate/clean_validation.json",
-        outputs=metadata.get("outputs", []),
-        ok=result.ok,
-        errors_count=len(result.errors),
-        warnings_count=len(result.warnings),
-    )
-    logger.info(f"VALIDATE CLEAN -> {report} (ok={result.ok})")
+    logger.info(f"VALIDATE CLEAN -> (ok={result.ok})")
     return build_validation_summary(result)

toolkit/cli/cmd_run.py

@@ -315,6 +315,8 @@ def run_year(
     run_has_validation_warnings = False
     sample_mode = sample_rows is not None or sample_bytes is not None
 
+    validations: dict[str, dict[str, Any]] = {}
+
     def _execute_layer(layer_name: str, target, *args, **kwargs) -> bool:
         """Esegue un layer e restituisce True se ok, False se fallito.
 
@@ -326,6 +328,7 @@ def _execute_layer(layer_name: str, target, *args, **kwargs) -> bool:
           warn_only: gli errori di validazione diventano warning, la pipeline continua
         """
         nonlocal run_has_validation_warnings
+        nonlocal validations
 
         layer_logger = bind_logger(base_logger, layer=layer_name)
         context.start_layer(layer_name)
@@ -340,6 +343,7 @@ def _execute_layer(layer_name: str, target, *args, **kwargs) -> bool:
                 validation_kwargs["sample_mode"] = sample_mode
             summary = _validation_runner(layer_name)(cfg, year, layer_logger, **validation_kwargs)
             context.set_validation(layer_name, summary)
+            validations[layer_name] = summary
             if not summary.get("passed", False):
                 message = f"{layer_name.upper()} validation failed"
                 if validation_mode == "strict" and fail_on_error:
@@ -962,6 +966,7 @@ def run_full(
                 results["steps"][str(year)] = {
                     "run": "ok",
                     "validate": "passed" if all_passed else "failed",
+                    "validations": ctx.validations,
                 }
                 if not all_passed and fail_on_error_flag:
                     results["status"] = "failed"

toolkit/core/metadata.py

@@ -84,36 +84,21 @@ def read_layer_metadata(layer_dir: Path) -> dict[str, Any]:
 
 
 # ---------------------------------------------------------------------------
-# merge_layer_manifest — merge validation fields into metadata.json
+# merge_layer_manifest — merge output file info into metadata.json
 # ---------------------------------------------------------------------------
 
 
 def merge_layer_manifest(
     folder: Path,
     *,
     metadata_path: str = METADATA,
-    validation_path: str | None = None,
     outputs: list[dict[str, Any]] | None = None,
-    ok: bool | None = None,
-    errors_count: int | None = None,
-    warnings_count: int | None = None,
     primary_output_file: str | None = None,
     sources: list[Any] | None = None,
 ) -> Path:
     meta = _read_metadata(folder, metadata_path) or {}
     if outputs is not None:
         meta["outputs"] = outputs
-    if validation_path is not None:
-        meta["validation"] = validation_path
-    if ok is not None or errors_count is not None or warnings_count is not None:
-        summary: dict[str, Any] = {}
-        if ok is not None:
-            summary["ok"] = ok
-        if errors_count is not None:
-            summary["errors_count"] = errors_count
-        if warnings_count is not None:
-            summary["warnings_count"] = warnings_count
-        meta["summary"] = summary
     if primary_output_file and not meta.get("primary_output_file"):
         meta["primary_output_file"] = primary_output_file
     if sources and not meta.get("sources"):

toolkit/core/validation.py

@@ -106,6 +106,13 @@ def build_validation_summary(result: ValidationResult) -> dict[str, Any]:
     }
     if "stats" in result.summary:
         out["stats"] = result.summary["stats"]
+    if result.sections:
+        out["sections"] = {}
+        for key, section in result.sections.items():
+            if isinstance(section, dict):
+                out["sections"][key] = {k: v for k, v in section.items() if k != "warning_messages"}
+            else:
+                out["sections"][key] = section
     return out
 
 

toolkit/domain/report.py

@@ -13,12 +13,9 @@
 from toolkit.core.io import read_json_or_none, write_json_atomic
 from toolkit.core.metadata import read_layer_metadata
 from toolkit.core.paths import (
-    CLEAN_VALIDATION,
-    MART_VALIDATION,
     METADATA,
     RAW_PROFILE,
     RAW_PROFILE_DIR,
-    RAW_VALIDATION,
     layer_year_dir,
 )
 from toolkit.core.run_records import get_run_dir, latest_run
@@ -28,38 +25,6 @@
 _DATASET_README = "README.md"
 
 
-# ---------------------------------------------------------------------------
-# Helpers
-# ---------------------------------------------------------------------------
-
-
-def _read_validation(root: Path, layer: str, dataset: str, year: int) -> dict[str, Any]:
-    """Legge il validation JSON di un layer, restituendo dict vuoto se non presente."""
-    val_map = {
-        "raw": RAW_VALIDATION,
-        "clean": CLEAN_VALIDATION,
-        "mart": MART_VALIDATION,
-    }
-    val_name = val_map.get(layer)
-    if not val_name:
-        return {}
-    val_path = layer_year_dir(root, layer, dataset, year) / val_name
-    if val_path.exists():
-        return read_json_or_none(val_path) or {}
-    val_path2 = layer_year_dir(root, layer, dataset, year) / "_validate" / val_name
-    if val_path2.exists():
-        return read_json_or_none(val_path2) or {}
-    return {}
-
-
-def _get_warnings(validation: dict[str, Any]) -> list[str]:
-    return validation.get("warnings", [])
-
-
-def _get_errors(validation: dict[str, Any]) -> list[str]:
-    return validation.get("errors", [])
-
-
 def _get_run_record(root: Path, dataset: str, year: int) -> dict[str, Any] | None:
     """Legge il run record più recente per un dataset/anno."""
     run_dir = get_run_dir(root, dataset, year)
@@ -304,17 +269,18 @@ def build_run_report(
     # --- Config hash (da metadata di qualsiasi layer) ---
     config_hash = _collect_config_hash(root_path, dataset, year)
 
-    # --- Validation per layer ---
+    # --- Validation per layer (da memoria, non da disco) ---
+    step_val = (step_results or {}).get("validations") or {}
     layers_report: dict[str, Any] = {}
     for lname in ("raw", "clean", "mart"):
-        val = _read_validation(root_path, lname, dataset, year)
-        warnings = _get_warnings(val)
-        errors = _get_errors(val)
+        val = step_val.get(lname, {})
+        warnings = val.get("warnings", [])
+        errors = val.get("errors", [])
         file_bytes = _collect_output_bytes(root_path, lname, dataset, year)
         layer_status = ((record or {}).get("layers") or {}).get(lname, {}).get("status")
 
         # Distingue layer non presente (es. mart-only: raw/clean assenti) da layer fallito
-        val_ok: bool | None = val.get("ok")
+        val_ok = val.get("passed")
         if val_ok is None and not val and layer_status is None:
             val_ok = None  # layer non eseguito (non è un fallimento)
 

toolkit/mart/run.py

@@ -22,7 +22,6 @@
 )
 from toolkit.core.multi_year_source import bind_multi_year_view, collect_multi_year_files
 from toolkit.core.paths import (
-    MART_VALIDATION,
     layer_dataset_dir,
     layer_year_dir,
     resolve_root,
@@ -498,7 +497,6 @@ def run_mart(
     merge_layer_manifest(
         mart_dir,
         metadata_path=metadata_path.name,
-        validation_path=MART_VALIDATION,
         outputs=outputs,
     )
     total_bytes = sum(p.stat().st_size for p in written if p.exists())

toolkit/mart/validate.py

@@ -13,15 +13,13 @@
 )
 from toolkit.core.io import read_json_or_none
 from toolkit.core.config import MartTableRuleConfig, MartValidationSpec
-from toolkit.core.metadata import merge_layer_manifest
-from toolkit.core.paths import MART_VALIDATION, METADATA, layer_year_dir, to_root_relative
+from toolkit.core.paths import METADATA, layer_year_dir, to_root_relative
 from toolkit.core.sql_utils import sql_path
 from toolkit.core.validation import (
     ValidationResult,
     build_validation_summary,
     check_transitions,
     required_columns_check,
-    write_validation_json,
 )
 
 
@@ -240,15 +238,5 @@ def run_mart_validation(cfg, year: int, logger, *, sample_mode: bool = False) ->
             sections={"transition": transition_report},
         )
 
-    report = write_validation_json(Path(mart_dir) / MART_VALIDATION, result)
-    merge_layer_manifest(
-        mart_dir,
-        metadata_path=METADATA,
-        validation_path="_validate/mart_validation.json",
-        outputs=metadata.get("outputs", []),
-        ok=result.ok,
-        errors_count=len(result.errors),
-        warnings_count=len(result.warnings),
-    )
-    logger.info(f"VALIDATE MART -> {report} (ok={result.ok})")
+    logger.info(f"VALIDATE MART -> (ok={result.ok})")
     return build_validation_summary(result)