Record verifiable demo run manifests

README.md

@@ -105,6 +105,11 @@ existing ignored `data/`, `reports/`, and `models/` directories. Use
 form, `python -m src.demo_pipeline`, remains available when working directly
 from a source checkout.
 
+Each successful run also writes `reports/sample_run_manifest.json`. The
+manifest records the effective pipeline parameters, Python and dependency
+versions, and the byte size and SHA-256 checksum of every generated artifact.
+This makes a saved demo run self-describing and independently verifiable.
+
 The equivalent installed console commands are:
 
 ```powershell

src/demo_pipeline.py

@@ -3,7 +3,11 @@
 from __future__ import annotations
 
 import argparse
+import hashlib
+import json
+import platform
 from dataclasses import dataclass
+from importlib.metadata import PackageNotFoundError, version
 from pathlib import Path
 
 import joblib
@@ -35,6 +39,16 @@
 DEFAULT_OUTPUT_DIR = Path(".")
 DEFAULT_HORIZON = 24
 LAG_HISTORY_HOURS = 168
+MANIFEST_SCHEMA_VERSION = 1
+RUNTIME_DISTRIBUTIONS = (
+    "aep-load-forecasting",
+    "joblib",
+    "matplotlib",
+    "numpy",
+    "pandas",
+    "scikit-learn",
+    "xgboost",
+)
 
 
 @dataclass(frozen=True)
@@ -50,9 +64,10 @@ class DemoArtifacts:
     model: Path
     forecast: Path
     forecast_plot: Path
+    manifest: Path
 
-    def paths(self) -> tuple[Path, ...]:
-        """Return every generated artifact in pipeline order."""
+    def output_paths(self) -> tuple[Path, ...]:
+        """Return generated data, report, model, and plot paths."""
 
         return (
             self.source,
@@ -66,6 +81,11 @@ def paths(self) -> tuple[Path, ...]:
             self.forecast_plot,
         )
 
+    def paths(self) -> tuple[Path, ...]:
+        """Return every generated artifact in pipeline order."""
+
+        return (*self.output_paths(), self.manifest)
+
 
 def demo_artifacts(output_dir: str | Path) -> DemoArtifacts:
     """Build the deterministic artifact layout below one output directory."""
@@ -87,7 +107,64 @@ def demo_artifacts(output_dir: str | Path) -> DemoArtifacts:
         forecast_plot=(
             root / "reports" / "figures" / "sample_forecast.png"
         ),
+        manifest=root / "reports" / "sample_run_manifest.json",
+    )
+
+
+def _sha256(path: Path) -> str:
+    """Return the SHA-256 digest of one artifact without loading it at once."""
+
+    digest = hashlib.sha256()
+    with path.open("rb") as file:
+        for chunk in iter(lambda: file.read(1024 * 1024), b""):
+            digest.update(chunk)
+    return digest.hexdigest()
+
+
+def _runtime_versions() -> dict[str, object]:
+    """Return the Python and package versions needed to reproduce a run."""
+
+    packages: dict[str, str] = {}
+    for distribution in RUNTIME_DISTRIBUTIONS:
+        try:
+            packages[distribution] = version(distribution)
+        except PackageNotFoundError:
+            packages[distribution] = "not-installed"
+    return {
+        "python": platform.python_version(),
+        "packages": packages,
+    }
+
+
+def save_run_manifest(
+    artifacts: DemoArtifacts,
+    output_dir: str | Path,
+    *,
+    parameters: dict[str, int | str],
+) -> Path:
+    """Record effective settings, runtime versions, and artifact checksums."""
+
+    root = Path(output_dir)
+    artifact_records: dict[str, dict[str, int | str]] = {}
+    for artifact in artifacts.output_paths():
+        relative_path = artifact.relative_to(root).as_posix()
+        artifact_records[relative_path] = {
+            "bytes": artifact.stat().st_size,
+            "sha256": _sha256(artifact),
+        }
+
+    payload = {
+        "schema_version": MANIFEST_SCHEMA_VERSION,
+        "parameters": parameters,
+        "runtime": _runtime_versions(),
+        "artifacts": artifact_records,
+    }
+    artifacts.manifest.parent.mkdir(parents=True, exist_ok=True)
+    artifacts.manifest.write_text(
+        json.dumps(payload, indent=2, sort_keys=True) + "\n",
+        encoding="utf-8",
     )
+    return artifacts.manifest
 
 
 def _validate_run_settings(
@@ -202,6 +279,19 @@ def run_demo_pipeline(
         forecast,
         artifacts.forecast_plot,
     )
+    save_run_manifest(
+        artifacts,
+        output_dir,
+        parameters={
+            "days": days,
+            "start": start,
+            "seed": seed,
+            "evaluation_days": evaluation_days,
+            "plot_days": plot_days,
+            "horizon": horizon,
+            "n_estimators": n_estimators,
+        },
+    )
     return artifacts
 
 

tests/test_demo_pipeline.py

@@ -1,5 +1,8 @@
 from __future__ import annotations
 
+import hashlib
+import json
+
 import joblib
 import pandas as pd
 import pytest
@@ -39,6 +42,7 @@ def test_command_writes_complete_reproducible_demo(tmp_path) -> None:
         output_dir / "models" / "sample_xgb.joblib",
         output_dir / "reports" / "sample_forecast.csv",
         output_dir / "reports" / "figures" / "sample_forecast.png",
+        output_dir / "reports" / "sample_run_manifest.json",
     ]
     assert all(path.is_file() for path in expected)
 
@@ -60,6 +64,32 @@ def test_command_writes_complete_reproducible_demo(tmp_path) -> None:
         "baseline_blend_MW",
     }.issubset(forecast.columns)
 
+    manifest = json.loads(expected[9].read_text(encoding="utf-8"))
+    assert manifest["schema_version"] == 1
+    assert manifest["parameters"] == {
+        "days": 13,
+        "start": "2025-01-01 00:00:00",
+        "seed": 42,
+        "evaluation_days": 2,
+        "plot_days": 1,
+        "horizon": 3,
+        "n_estimators": 5,
+    }
+    assert manifest["runtime"]["python"]
+    assert manifest["runtime"]["packages"]["xgboost"]
+
+    artifact_records = manifest["artifacts"]
+    expected_artifacts = expected[:9]
+    assert set(artifact_records) == {
+        path.relative_to(output_dir).as_posix()
+        for path in expected_artifacts
+    }
+    for path in expected_artifacts:
+        record = artifact_records[path.relative_to(output_dir).as_posix()]
+        contents = path.read_bytes()
+        assert record["bytes"] == len(contents)
+        assert record["sha256"] == hashlib.sha256(contents).hexdigest()
+
 
 def test_pipeline_rejects_too_little_history_before_writing(tmp_path) -> None:
     output_dir = tmp_path / "demo"