feat: Lattice.save/load — accumulator snapshot persistence

CLAUDE.md

@@ -48,10 +48,11 @@ no hit-policy interaction. Shares Steps 1–3 with `evaluate` via
 - `apply(lattice, context)` → `AccumulatorResult`. Apply-phase filter engines are memoised per lattice (WeakKeyDictionary).
 - `DimensionRole.CONTEXT_KEY` dimensions partition the rule space; `build_all` + `index(lattices)` → `LatticeIndex` routes contexts (single or batch) to the right lattice.
 - `Lattice.is_composed` distinguishes build output (has `__prime_product`) from flat/imported lattices.
+- `Lattice.save(dir)` / `Lattice.load(dir)` — snapshot persistence (`lattice.parquet` + `manifest.yaml`, a superset of babel's LatticeManifest). `load` carries the package's third `# allow:` tag (parquet read). Build offline, `save`, serve `apply` from `load`.
 
 ### Backend purity (ENFORCED)
 
-No module under `src/mountainash_rules/` may import polars/ibis/narwhals directly — only `mountainash.relations` / `mountainash.expressions`. `tests/test_backend_purity.py` enforces this across the whole package (every non-dunder module is parametrised, no exemptions). A genuinely unavoidable native escape must be tagged `# allow: <reason>` on the import line — currently two: the per-row REGEX fallback in `core/compiler.py` (pending upstream column-pattern `regex_contains`) and the empty-build schema seed in `engines/accumulator/engine.py` (pending backend-agnostic empty-frame support).
+No module under `src/mountainash_rules/` may import polars/ibis/narwhals directly — only `mountainash.relations` / `mountainash.expressions`. `tests/test_backend_purity.py` enforces this across the whole package (every non-dunder module is parametrised, no exemptions). A genuinely unavoidable native escape must be tagged `# allow: <reason>` on the import line — currently three: the per-row REGEX fallback in `core/compiler.py` (pending upstream column-pattern `regex_contains`), the empty-build schema seed in `engines/accumulator/engine.py` (pending backend-agnostic empty-frame support), and the snapshot parquet read in `engines/accumulator/lattice.py` (pending backend-agnostic `parquet.read_table`).
 
 ## Match Strategies
 

src/mountainash_rules/engines/accumulator/lattice.py

@@ -1,4 +1,6 @@
+import pathlib
 import typing as t
+import yaml
 
 from mountainash.relations import relation
 
@@ -54,6 +56,49 @@ def metadata(self) -> DimensionsMetadata:
     def aggregates(self) -> list[Aggregate]:
         return self._aggregates
 
+    def save(self, dir_path: "str | pathlib.Path") -> "pathlib.Path":
+        """Persist this lattice as a snapshot directory (parquet + manifest).
+
+        The manifest is a strict superset of babel's LatticeManifest YAML,
+        so babel/service tooling can read it unchanged.
+        """
+        dir_path = pathlib.Path(dir_path)
+        dir_path.mkdir(parents=True, exist_ok=True)
+        relation(self._df).to_polars().write_parquet(dir_path / "lattice.parquet")
+        manifest = {
+            "dimensions": self._metadata.model_dump(
+                mode="json", exclude_defaults=True
+            ),
+            "aggregates": [
+                a.model_dump(mode="json") for a in self._aggregates
+            ],
+            "partition_key": self._partition_key,
+        }
+        (dir_path / "manifest.yaml").write_text(
+            yaml.safe_dump(manifest, sort_keys=False), encoding="utf-8"
+        )
+        return dir_path
+
+    @classmethod
+    def load(cls, dir_path: "str | pathlib.Path") -> "Lattice":
+        """Rehydrate a snapshot written by save(). Preserves __agg_* and
+        __prime_product verbatim (is_composed round-trips honestly)."""
+        import polars as pl  # allow: lattice snapshot parquet read pending backend-agnostic file IO
+
+        dir_path = pathlib.Path(dir_path)
+        manifest_path = dir_path / "manifest.yaml"
+        if not manifest_path.exists():
+            raise FileNotFoundError(f"No manifest.yaml in {dir_path}")
+        raw = yaml.safe_load(manifest_path.read_text(encoding="utf-8"))
+        return cls(
+            dataframe=pl.read_parquet(dir_path / "lattice.parquet"),
+            metadata=DimensionsMetadata.model_validate(raw["dimensions"]),
+            aggregates=[
+                Aggregate.model_validate(a) for a in raw.get("aggregates", [])
+            ],
+            partition_key=raw.get("partition_key"),
+        )
+
 
 class LatticeIndex:
     """Partition-key routing over a set of built lattices, built once."""

tests/accumulator/test_lattice.py

@@ -93,3 +93,83 @@ def test_empty_build_is_still_composed(self):
         assert lattice.count == 0
         assert lattice.is_composed is True
         assert "co_region" in relation(lattice.combinations).columns
+
+
+from mountainash_rules import (
+    AccumulatorEngine,
+    Aggregate,
+    Dimension,
+    DimensionsMetadata,
+    Lattice,
+    MatchStrategy,
+    UNKNOWN,
+)
+
+
+@pytest.fixture
+def built_lattice_and_engine():
+    import polars as pl
+
+    metadata = DimensionsMetadata(
+        dimensions=[
+            Dimension(dimension_name="region", match_strategy=MatchStrategy.EXACT, data_type="str"),
+            Dimension(dimension_name="channel", match_strategy=MatchStrategy.EXACT, data_type="str"),
+        ]
+    )
+    rules = pl.DataFrame(
+        {
+            "rule_name": ["au_base", "broker_bonus", "au_broker"],
+            "region": ["AU", UNKNOWN, "AU"],
+            "channel": [UNKNOWN, "BROKER", "BROKER"],
+            "discount": [5.0, 2.5, 10.0],
+        }
+    )
+    engine = AccumulatorEngine(
+        dimension_metadata=metadata,
+        aggregates=[Aggregate(column_name="discount", operation="sum")],
+    )
+    return engine.build(rules), engine
+
+
+class TestLatticeSaveLoad:
+    def test_round_trip_identity(self, built_lattice_and_engine, tmp_path):
+        lattice, _ = built_lattice_and_engine
+        out = lattice.save(tmp_path / "snap")
+        loaded = Lattice.load(out)
+        assert loaded.count == lattice.count
+        assert loaded.is_composed is True  # __prime_product travels
+        assert loaded.metadata == lattice.metadata
+        assert loaded.aggregates == lattice.aggregates
+        assert loaded.partition_key == lattice.partition_key
+
+    def test_apply_equivalence(self, built_lattice_and_engine, tmp_path):
+        lattice, engine = built_lattice_and_engine
+        loaded = Lattice.load(lattice.save(tmp_path / "snap"))
+        ctx = {"region": "AU", "channel": "BROKER"}
+        original = engine.apply(lattice, ctx)
+        reloaded = engine.apply(loaded, ctx)
+        assert reloaded.count == original.count
+        assert (
+            reloaded.accumulated("discount").to_dicts()
+            == original.accumulated("discount").to_dicts()
+        )
+
+    def test_save_creates_expected_files(self, built_lattice_and_engine, tmp_path):
+        lattice, _ = built_lattice_and_engine
+        out = lattice.save(tmp_path / "snap")
+        assert (out / "lattice.parquet").exists()
+        assert (out / "manifest.yaml").exists()
+
+    def test_manifest_is_babel_superset(self, built_lattice_and_engine, tmp_path):
+        import yaml
+
+        lattice, _ = built_lattice_and_engine
+        out = lattice.save(tmp_path / "snap")
+        raw = yaml.safe_load((out / "manifest.yaml").read_text())
+        assert set(raw) == {"dimensions", "aggregates", "partition_key"}
+        assert raw["aggregates"] == [{"column_name": "discount", "operation": "sum"}]
+
+    def test_load_missing_manifest_raises(self, tmp_path):
+        (tmp_path / "empty").mkdir()
+        with pytest.raises(FileNotFoundError):
+            Lattice.load(tmp_path / "empty")