From f306f5e3a4afbabf7b1b1485dc41f4b936b09f28 Mon Sep 17 00:00:00 2001 From: Nathaniel Ramm Date: Sun, 19 Jul 2026 18:44:17 +1000 Subject: [PATCH 01/11] docs: amend ternary-partition-routing spec (post-Codex review) + implementation plan Co-Authored-By: Claude Opus 4.8 (1M context) --- .../2026-07-19-ternary-partition-routing.md | 1067 +++++++++++++++++ ...-07-19-ternary-partition-routing-design.md | 160 ++- 2 files changed, 1200 insertions(+), 27 deletions(-) create mode 100644 docs/superpowers/plans/2026-07-19-ternary-partition-routing.md diff --git a/docs/superpowers/plans/2026-07-19-ternary-partition-routing.md b/docs/superpowers/plans/2026-07-19-ternary-partition-routing.md new file mode 100644 index 0000000..1c82ccb --- /dev/null +++ b/docs/superpowers/plans/2026-07-19-ternary-partition-routing.md @@ -0,0 +1,1067 @@ +# Ternary Partition Routing Implementation Plan + +> **For agentic workers:** REQUIRED SUB-SKILL: Use superpowers:subagent-driven-development (recommended) or superpowers:executing-plans to implement this plan task-by-task. Steps use checkbox (`- [ ]`) syntax for tracking. + +**Goal:** Generalise `LatticeIndex` partition routing from exact dict lookup to ternary + specificity semantics (wildcard/default partitions, load-time ambiguity validation), implemented by an embedded `ExpressionRulesEngine` over a meta rules-table. + +**Architecture:** A new `MatchStrategy.EXACT_KEY` (rule-side-wildcard-only exact match) gives routing its asymmetric ternary semantics. `LatticeIndex` builds a meta rules-table (one row per partition) evaluated by an internal filter engine; exact dict hits keep the O(1) fast path; `index()` validates the suite via an exhaustive witness-context matrix routed through the same meta-engine. + +**Tech Stack:** Python 3.12, pydantic, mountainash expressions/relations (backend-agnostic), polars in tests only, pytest via hatch. + +**Spec:** `docs/superpowers/specs/2026-07-19-ternary-partition-routing-design.md` (read it before starting — it is the authority on semantics). + +## Global Constraints + +- Branch: `feature/ternary-partition-routing` off `develop` @ `496f84a` (spec header). Create it first if it doesn't exist: `git checkout develop && git checkout -b feature/ternary-partition-routing`. +- **Backend purity (ENFORCED by `tests/test_backend_purity.py`):** no module under `src/mountainash_rules/` may import polars/ibis/narwhals directly. This plan needs **no new `# allow:` tag** — the purity count stays at three. +- Ternary encoding everywhere: 1 = match, 0 = wildcard/unknown, −1 = non-match; survive on `min ≥ 0`; specificity = count of 1s. +- Routing semantics are **asymmetric**: only the rule side (partition key) has a wildcard (UNKNOWN sentinel; `null` for bool). Context-side sentinels (NOT_SET, UNKNOWN, `None`) score −1 against specific keys, 0 against wildcards. +- Partition keys may **not** contain NOT_SET sentinels (`ValueError` at `index()`, always-on). +- `AmbiguousPartitionError` subclasses `KeyError` (deliberate — spec §1/§9). +- No-survivor routing stays a plain `KeyError` whose message starts `"No lattice"` (existing tests match on that prefix). +- TDD: failing test first, then implementation. Test commands: `hatch run test:test-target ::::` (single), `hatch run test:test-quick` (suite). Lint: `hatch run ruff:check`. Types: `hatch run mypy:check`. +- Style: `import typing as t`, Google docstrings, ruff-formatted. Match surrounding code. +- Do not modify `Lattice`, `Lattice.save/load`, `build`/`build_all`, or any service code. + +## File Structure + +| File | Responsibility | +|---|---| +| `src/mountainash_rules/core/constants.py` | add `MatchStrategy.EXACT_KEY` member | +| `src/mountainash_rules/core/compiler.py` | add `_compile_exact_key` + dispatch case | +| `src/mountainash_rules/engines/accumulator/engine.py` | `_extract_partition_key` NOT_SET fill; `_normalize_partition_key`; `index()` gains `validate`/`max_witnesses` | +| `src/mountainash_rules/engines/accumulator/lattice.py` | `AmbiguousPartitionError`; `LatticeIndex` structural checks, meta-engine, `_route`, witness-matrix validation, `apply`/`apply_batch` routing | +| `src/mountainash_rules/__init__.py` | export `AmbiguousPartitionError` | +| `tests/core/test_compiler.py` | `TestExactKeyCompilation` | +| `tests/accumulator/test_apply.py` | `_extract_partition_key` fill tests | +| `tests/accumulator/test_lattice.py` | `TestTernaryRouting`, `TestIndexValidation`, persistence round-trip | +| `tests/filter/test_batch_evaluation.py` | batch routing tests (existing index-batch tests live here) | +| `CLAUDE.md` | strategy table + accumulator section updates | + +--- + +### Task 1: `MatchStrategy.EXACT_KEY` — compiler strategy + +**Files:** +- Modify: `src/mountainash_rules/core/constants.py` (MatchStrategy enum, ~line 10) +- Modify: `src/mountainash_rules/core/compiler.py` (dispatch `match` block ~line 40, new method after `_compile_exact`) +- Test: `tests/core/test_compiler.py` + +**Interfaces:** +- Consumes: existing `DimensionCompiler.compile_dimension`, `unknown_sentinel_for`, `CTX_PREFIX`. +- Produces: `MatchStrategy.EXACT_KEY = "exact_key"`; `DimensionCompiler._compile_exact_key(dim) -> BaseExpressionAPI`. Semantics: rule value == UNKNOWN sentinel → 0; rule == context → 1; otherwise (including context-side NOT_SET/UNKNOWN/null) → −1. Bool: rule null → 0; else eq → 1; else −1. + +- [ ] **Step 1: Write the failing tests** + +Append to `tests/core/test_compiler.py` (imports of `NOT_SET` may need adding to the existing `from mountainash_rules.core.constants import ...` line): + +```python +class TestExactKeyCompilation: + """EXACT_KEY: rule-side wildcard only — context sentinels are non-matches.""" + + def test_rule_unknown_is_wildcard(self, compiler): + dim = Dimension( + dimension_name="region", + match_strategy=MatchStrategy.EXACT_KEY, + data_type="str", + ) + expr = compiler.compile_dimension(dim) + df = pl.DataFrame({ + "region": ["AU", UNKNOWN, "UK"], + f"{CTX_PREFIX}region": ["AU", "AU", "AU"], + }) + result = df.with_columns(expr.name.alias("__t_region").compile(df, booleanizer=None)) + assert result["__t_region"].to_list() == [1, 0, -1] + + def test_context_not_set_never_matches_specific(self, compiler): + # The asymmetry that distinguishes EXACT_KEY from EXACT: + # a NOT_SET context is -1 against specific keys (EXACT gives 0). + dim = Dimension( + dimension_name="region", + match_strategy=MatchStrategy.EXACT_KEY, + data_type="str", + ) + expr = compiler.compile_dimension(dim) + df = pl.DataFrame({ + "region": ["AU", UNKNOWN], + f"{CTX_PREFIX}region": [NOT_SET, NOT_SET], + }) + result = df.with_columns(expr.name.alias("__t_region").compile(df, booleanizer=None)) + assert result["__t_region"].to_list() == [-1, 0] + + def test_context_unknown_never_matches_specific(self, compiler): + dim = Dimension( + dimension_name="region", + match_strategy=MatchStrategy.EXACT_KEY, + data_type="str", + ) + expr = compiler.compile_dimension(dim) + df = pl.DataFrame({ + "region": ["AU", UNKNOWN], + f"{CTX_PREFIX}region": [UNKNOWN, UNKNOWN], + }) + result = df.with_columns(expr.name.alias("__t_region").compile(df, booleanizer=None)) + assert result["__t_region"].to_list() == [-1, 0] + + def test_numeric_sentinels(self, compiler): + dim = Dimension( + dimension_name="product_id", + match_strategy=MatchStrategy.EXACT_KEY, + data_type="int", + ) + expr = compiler.compile_dimension(dim) + df = pl.DataFrame({ + "product_id": [1, UNKNOWN_NUMERIC, 2], + f"{CTX_PREFIX}product_id": [1, 1, 1], + }) + result = df.with_columns(expr.name.alias("__t_product_id").compile(df, booleanizer=None)) + assert result["__t_product_id"].to_list() == [1, 0, -1] + + def test_bool_rule_null_is_wildcard_context_null_is_not(self, compiler): + dim = Dimension( + dimension_name="flag", + match_strategy=MatchStrategy.EXACT_KEY, + data_type="bool", + ) + expr = compiler.compile_dimension(dim) + df = pl.DataFrame({ + "flag": [True, None, True, None], + f"{CTX_PREFIX}flag": [True, True, None, None], + }) + result = df.with_columns(expr.name.alias("__t_flag").compile(df, booleanizer=None)) + # rule null -> 0 regardless of context; context null vs specific -> -1 + assert result["__t_flag"].to_list() == [1, 0, -1, 0] +``` + +`NOT_SET` must be added to the constants import at the top of the file if not present. + +- [ ] **Step 2: Run tests to verify they fail** + +Run: `hatch run test:test-target tests/core/test_compiler.py::TestExactKeyCompilation -v` +Expected: FAIL — `AttributeError: EXACT_KEY` (enum member doesn't exist). + +- [ ] **Step 3: Implement** + +In `src/mountainash_rules/core/constants.py`, add to `MatchStrategy` after `NOT_EQUAL`: + +```python + EXACT_KEY = "exact_key" +``` + +In `src/mountainash_rules/core/compiler.py`: + +Add a dispatch case in `compile_dimension`'s `match` block, next to the EXACT case: + +```python + case MatchStrategy.EXACT_KEY: + return self._compile_exact_key(dim) +``` + +Add the method directly after `_compile_exact` (it needs `unknown_sentinel_for` added to the existing `constants` import): + +```python + def _compile_exact_key(self, dim: Dimension) -> BaseExpressionAPI: + """Rule-side-wildcard-only exact match (partition-key routing). + + Only the rule side has a wildcard (UNKNOWN sentinel; null for + bool). A context-side sentinel is an ordinary non-matching value: + specific keys must never match an unknown/unset context. + """ + rule_col = ma.col(dim.resolved_rule_field) + ctx_col = ma.col(CTX_PREFIX + dim.dimension_name) + if dim.data_type is DataType.BOOL: + wildcard = rule_col.is_null() + else: + wildcard = rule_col.__eq__( + ma.lit(unknown_sentinel_for(dim.data_type)) + ) + return ( + ma.when(wildcard).then(0) + .when(rule_col.__eq__(ctx_col)).then(1) + .otherwise(-1) + ) +``` + +(A null comparison result falls through `when` to `otherwise(-1)`, which is exactly the required context-null behaviour — same pattern as the existing string-strategy sentinel wrapper at `_wrap`.) + +No `dimension.py` validation changes: `EXACT_KEY` accepts every data type and has no strategy-specific fields, like `EXACT`. + +- [ ] **Step 4: Run tests to verify they pass** + +Run: `hatch run test:test-target tests/core/test_compiler.py::TestExactKeyCompilation -v` +Expected: 5 PASS. + +- [ ] **Step 5: Run the quick suite, lint, and commit** + +Run: `hatch run test:test-quick && hatch run ruff:check` +Expected: all pass (the parametrised purity and strategy tests must stay green). + +```bash +git add src/mountainash_rules/core/constants.py src/mountainash_rules/core/compiler.py tests/core/test_compiler.py +git commit -m "feat: add MatchStrategy.EXACT_KEY — rule-side-wildcard-only exact match" +``` + +--- + +### Task 2: `_extract_partition_key` — NOT_SET fill for missing/null fields + +**Files:** +- Modify: `src/mountainash_rules/engines/accumulator/engine.py:548-565` (`_extract_partition_key`) +- Test: `tests/accumulator/test_apply.py` + +**Interfaces:** +- Consumes: `not_set_sentinel_for` from `core.constants` (already importable), `DataType`. +- Produces: `AccumulatorEngine._extract_partition_key(context) -> tuple` — missing **or explicitly-None** context field becomes `not_set_sentinel_for(d.data_type)` (`None` for bool) instead of raising `KeyError`. Also `AccumulatorEngine._normalize_partition_key(key: tuple) -> tuple` applying the same None→sentinel rule to an already-extracted tuple (used by batch routing in Task 4). + +- [ ] **Step 1: Write the failing tests** + +Append to `tests/accumulator/test_apply.py` (add `NOT_SET_NUMERIC` to the constants import at the top): + +```python +class TestExtractPartitionKey: + def _engine(self): + metadata = DimensionsMetadata(dimensions=[ + Dimension( + dimension_name="product_id", + match_strategy=MatchStrategy.EXACT, + data_type=int, + role=DimensionRole.CONTEXT_KEY, + ), + Dimension(dimension_name="channel", match_strategy=MatchStrategy.EXACT), + ]) + return AccumulatorEngine( + dimension_metadata=metadata, + aggregates=[Aggregate(column_name="margin")], + ) + + def test_missing_key_field_fills_not_set(self): + key = self._engine()._extract_partition_key({"channel": "BROKER"}) + assert key == (NOT_SET_NUMERIC,) + + def test_explicit_none_fills_not_set(self): + key = self._engine()._extract_partition_key( + {"product_id": None, "channel": "BROKER"} + ) + assert key == (NOT_SET_NUMERIC,) + + def test_present_value_passes_through(self): + key = self._engine()._extract_partition_key( + {"product_id": 7, "channel": "BROKER"} + ) + assert key == (7,) + + def test_normalize_partition_key(self): + engine = self._engine() + assert engine._normalize_partition_key((None,)) == (NOT_SET_NUMERIC,) + assert engine._normalize_partition_key((7,)) == (7,) +``` + +- [ ] **Step 2: Run tests to verify they fail** + +Run: `hatch run test:test-target tests/accumulator/test_apply.py::TestExtractPartitionKey -v` +Expected: FAIL — `KeyError: 'product_id'` on the first two, `AttributeError: _normalize_partition_key` on the last. + +- [ ] **Step 3: Implement** + +Replace the body of `_extract_partition_key` in `src/mountainash_rules/engines/accumulator/engine.py` and add the helper. Ensure `DataType` and `not_set_sentinel_for` are in the module's `core.constants` import: + +```python + def _extract_partition_key(self, context: t.Any) -> tuple: + """Extract the partition key tuple from a context object. + + Missing or explicitly-null key fields become the typed NOT_SET + sentinel (None for bool) so the context can still route — a + NOT_SET value matches wildcard partitions only. + """ + if isinstance(context, BaseModel): + raw = context.model_dump() + elif isinstance(context, dict): + raw = context + else: + raise TypeError( + f"Context must be a BaseModel or dict, got {type(context).__name__}" + ) + return self._normalize_partition_key(tuple( + raw.get(d.resolved_context_field) + for d in self._context_key_dims + )) + + def _normalize_partition_key(self, key: tuple) -> tuple: + """Map None key values to the typed NOT_SET sentinel (None for bool).""" + return tuple( + v if v is not None or d.data_type is DataType.BOOL + else not_set_sentinel_for(d.data_type) + for v, d in zip(key, self._context_key_dims) + ) +``` + +- [ ] **Step 4: Run tests to verify they pass** + +Run: `hatch run test:test-target tests/accumulator/test_apply.py::TestExtractPartitionKey -v` +Expected: 4 PASS. Then `hatch run test:test-quick` — the existing `test_apply_auto_missing_key_raises` must still pass (context `product_id=99` is present, so it still misses via the dict → `KeyError: "No lattice..."`). + +- [ ] **Step 5: Commit** + +```bash +git add src/mountainash_rules/engines/accumulator/engine.py tests/accumulator/test_apply.py +git commit -m "feat: NOT_SET fill for missing/null partition key fields" +``` + +--- + +### Task 3: Meta-engine routing — `AmbiguousPartitionError`, structural checks, `_route`, `apply` + +**Files:** +- Modify: `src/mountainash_rules/engines/accumulator/lattice.py` (`LatticeIndex.__init__`, `apply`; new exception; new `_route`) +- Modify: `src/mountainash_rules/__init__.py` (export) +- Test: `tests/accumulator/test_lattice.py` + +**Interfaces:** +- Consumes: `MatchStrategy.EXACT_KEY` (Task 1), `engine._extract_partition_key` (Task 2), `ExpressionRulesEngine` (filter layer), `relation` (already imported in lattice.py). +- Produces: + - `AmbiguousPartitionError(KeyError)` in `lattice.py`, exported from package root. + - `LatticeIndex.__init__(engine, lattices, context_key_dims)` — raises `ValueError` on empty list, duplicate keys, NOT_SET-bearing keys, or a `partition_key=None` lattice when key dims exist; builds `self._lattices` (list), `self._map` (dict), `self._meta_engine` (or `None` when there are no key dims). + - `LatticeIndex._route(key: tuple) -> Lattice` — meta-engine evaluate; unique top-specificity survivor wins; 0 survivors → `KeyError` starting `"No lattice"` and listing served keys; ≥2 tied → `AmbiguousPartitionError`. + - `LatticeIndex.apply(context, dimensions=None)` — dict fast path, `_route` on miss. + - Validation (Task 5) reuses `self._meta_engine` and `self._lattices`. + +- [ ] **Step 1: Write the failing tests** + +Append to `tests/accumulator/test_lattice.py`. The file already imports `Lattice`; extend imports as shown: + +```python +from pydantic import BaseModel + +from mountainash_rules import AmbiguousPartitionError +from mountainash_rules.engines.accumulator.engine import AccumulatorEngine +from mountainash_rules.engines.accumulator.aggregate import Aggregate +from mountainash_rules.core.constants import ( + UNKNOWN, + UNKNOWN_NUMERIC, + NOT_SET, + DimensionRole, + MatchStrategy, +) +from mountainash_rules.core.dimension import Dimension, DimensionsMetadata + + +class RoutingContext(BaseModel): + region: str | None = None + channel: str | None = None + product: str + + +def _routing_engine(): + metadata = DimensionsMetadata(dimensions=[ + Dimension( + dimension_name="region", + match_strategy=MatchStrategy.EXACT, + role=DimensionRole.CONTEXT_KEY, + ), + Dimension( + dimension_name="channel", + match_strategy=MatchStrategy.EXACT, + role=DimensionRole.CONTEXT_KEY, + ), + Dimension(dimension_name="product", match_strategy=MatchStrategy.EXACT), + ]) + return AccumulatorEngine( + dimension_metadata=metadata, + aggregates=[Aggregate(column_name="margin")], + ) + + +def _routing_rules(keys: list[tuple[str, str]]) -> pl.DataFrame: + """One rule per (region, channel) partition key; UNKNOWN = wildcard.""" + return pl.DataFrame({ + "region": [k[0] for k in keys], + "channel": [k[1] for k in keys], + "rule_name": [f"r{i}" for i in range(len(keys))], + "product": ["GOLD"] * len(keys), + "margin": [1.0] * len(keys), + }) + + +def _index_for(keys, validate=True): + engine = _routing_engine() + lattices = engine.build_all(_routing_rules(keys)) + return engine, engine.index(lattices, validate=validate) + + +class TestTernaryRouting: + def test_default_partition_catches_unmatched_context(self): + engine, index = _index_for([("AU", "BROKER"), (UNKNOWN, UNKNOWN)]) + result = index.apply(RoutingContext(region="NZ", channel="DIRECT", product="GOLD")) + assert result.count >= 1 # routed to the all-wildcard default + + def test_partial_specific_miss_falls_to_default(self): + engine, index = _index_for([("AU", "BROKER"), (UNKNOWN, UNKNOWN)]) + # Exercise _route (not the dict fast path): (AU, DIRECT) is not an + # exact key; (AU, BROKER) dies on channel; default survives. + result = index.apply(RoutingContext(region="AU", channel="DIRECT", product="GOLD")) + assert result.count >= 1 + + def test_exact_hit_uses_fast_path_and_wins(self): + engine, index = _index_for([("AU", "BROKER"), (UNKNOWN, UNKNOWN)]) + result = index.apply(RoutingContext(region="AU", channel="BROKER", product="GOLD")) + assert result.count >= 1 + + def test_runtime_ambiguity_raises(self): + engine, index = _index_for( + [("AU", UNKNOWN), (UNKNOWN, "BROKER")], validate=False + ) + with pytest.raises(AmbiguousPartitionError): + index.apply(RoutingContext(region="AU", channel="BROKER", product="GOLD")) + + def test_missing_key_field_routes_to_default(self): + engine, index = _index_for([("AU", "BROKER"), (UNKNOWN, UNKNOWN)]) + result = index.apply(RoutingContext(product="GOLD")) # region/channel None + assert result.count >= 1 + + def test_missing_key_field_without_default_raises(self): + engine, index = _index_for([("AU", "BROKER")]) + with pytest.raises(KeyError, match="No lattice"): + index.apply(RoutingContext(product="GOLD")) + + def test_context_unknown_sentinel_is_wildcard_only(self): + engine, index = _index_for([("AU", "BROKER"), (UNKNOWN, UNKNOWN)]) + result = index.apply( + RoutingContext(region=UNKNOWN, channel="BROKER", product="GOLD") + ) + # UNKNOWN region kills (AU, BROKER); only the default survives. + assert result.count >= 1 + + def test_wildcard_free_index_miss_raises_keyerror(self): + engine, index = _index_for([("AU", "BROKER")]) + with pytest.raises(KeyError, match="No lattice"): + index.apply(RoutingContext(region="US", channel="X", product="GOLD")) + + def test_ambiguous_is_a_keyerror(self): + assert issubclass(AmbiguousPartitionError, KeyError) + + +class TestIndexStructuralChecks: + def test_empty_index_raises(self): + engine = _routing_engine() + with pytest.raises(ValueError, match="at least one lattice"): + engine.index([]) + + def test_duplicate_keys_raise(self): + engine = _routing_engine() + lattices = engine.build_all(_routing_rules([("AU", "BROKER")])) + with pytest.raises(ValueError, match="[Dd]uplicate"): + engine.index(lattices + lattices) + + def test_not_set_key_raises(self): + engine = _routing_engine() + lattices = engine.build_all(_routing_rules([(NOT_SET, "BROKER")])) + with pytest.raises(ValueError, match="NOT_SET"): + engine.index(lattices) + + def test_keyless_lattice_with_key_dims_raises(self): + engine = _routing_engine() + (good,) = engine.build_all(_routing_rules([("AU", "BROKER")])) + flat = Lattice( + dataframe=good.combinations, + metadata=good.metadata, + aggregates=good.aggregates, + partition_key=None, + ) + with pytest.raises(ValueError, match="partition_key"): + engine.index([good, flat]) +``` + +- [ ] **Step 2: Run tests to verify they fail** + +Run: `hatch run test:test-target tests/accumulator/test_lattice.py::TestTernaryRouting tests/accumulator/test_lattice.py::TestIndexStructuralChecks -v` +Expected: FAIL — `ImportError: cannot import name 'AmbiguousPartitionError'`. + +- [ ] **Step 3: Implement** + +In `src/mountainash_rules/engines/accumulator/lattice.py`: + +Extend the module imports: + +```python +from mountainash_rules.core.constants import ( + DataType, + DimensionRole, + HitPolicy, + MatchStrategy, + not_set_sentinel_for, + unknown_sentinel_for, +) +from mountainash_rules.core.dimension import Dimension +``` + +(`DimensionsMetadata` is already imported. Importing `ExpressionRulesEngine` at module top is fine — `engines/filter` only imports `core`, no cycle: `from mountainash_rules.engines.filter.engine import ExpressionRulesEngine`.) + +Add the exception above `class LatticeIndex`: + +```python +class AmbiguousPartitionError(KeyError): + """Two or more partitions tie at top specificity for a context. + + Subclasses KeyError so existing partition-miss handlers keep working; + reachable at runtime only when index() validation was opted out. + """ +``` + +Replace `LatticeIndex.__init__` and `apply`, and add `_route` and the meta-engine builder: + +```python +class LatticeIndex: + """Partition-key routing over a set of built lattices, built once. + + Routing uses the same ternary + specificity semantics as the + constraint layer, evaluated by an embedded ExpressionRulesEngine over + a meta rules-table (one row per partition). Exact key hits keep an + O(1) dict fast path. + """ + + def __init__(self, engine, lattices: list["Lattice"], context_key_dims) -> None: + self._engine = engine + self._context_key_dims = list(context_key_dims) + self._lattices = list(lattices) + if not self._lattices: + raise ValueError("index() requires at least one lattice") + + self._map: dict[tuple, Lattice] = {} + for lattice in self._lattices: + if lattice.partition_key is not None: + key = tuple( + lattice.partition_key[d.dimension_name] + for d in self._context_key_dims + ) + elif self._context_key_dims: + raise ValueError( + "Lattice without a partition_key cannot be indexed " + "alongside CONTEXT_KEY dimensions" + ) + else: + key = () + for d, v in zip(self._context_key_dims, key): + if ( + d.data_type is not DataType.BOOL + and v == not_set_sentinel_for(d.data_type) + ): + raise ValueError( + f"Partition key {key!r} contains the NOT_SET " + f"sentinel for dimension '{d.dimension_name}'; " + f"NOT_SET is a context-side sentinel and would " + f"collide with missing-field routing" + ) + if key in self._map: + raise ValueError(f"Duplicate partition key {key!r}") + self._map[key] = lattice + + self._meta_engine = ( + self._build_meta_engine() if self._context_key_dims else None + ) + + def _build_meta_engine(self) -> "ExpressionRulesEngine": + """One meta-rule row per partition; EXACT_KEY per key dim.""" + columns: dict[str, list] = { + d.dimension_name: [] for d in self._context_key_dims + } + columns["__partition_idx"] = [] + for idx, lattice in enumerate(self._lattices): + for d in self._context_key_dims: + columns[d.dimension_name].append( + lattice.partition_key[d.dimension_name] + ) + columns["__partition_idx"].append(idx) + meta_metadata = DimensionsMetadata( + dimensions=[ + Dimension( + dimension_name=d.dimension_name, + context_field=d.resolved_context_field, + match_strategy=MatchStrategy.EXACT_KEY, + data_type=d.data_type, + ) + for d in self._context_key_dims + ], + hit_policy=HitPolicy.COLLECT, + ) + return ExpressionRulesEngine( + rules=relation(columns).collect(), + dimension_metadata=meta_metadata, + ) + + def _route(self, key: tuple) -> "Lattice": + """Ternary + specificity routing for a normalised key tuple.""" + ctx = { + d.resolved_context_field: key[i] + for i, d in enumerate(self._context_key_dims) + } + rows = relation(self._meta_engine.evaluate(ctx).survivors).to_dict() + idxs = rows["__partition_idx"] + if not idxs: + raise KeyError( + f"No lattice for partition key {key!r}; served partition " + f"keys: {sorted(self._map)!r}" + ) + top = rows["__specificity"][0] # survivors are rank-sorted + tied = [ + i for i, s in zip(idxs, rows["__specificity"]) if s == top + ] + if len(tied) > 1: + tied_keys = [ + tuple( + self._lattices[i].partition_key[d.dimension_name] + for d in self._context_key_dims + ) + for i in tied + ] + raise AmbiguousPartitionError( + f"Context key {key!r} ties {len(tied)} partitions at " + f"specificity {top}: {tied_keys!r}" + ) + return self._lattices[tied[0]] + + def apply(self, context, dimensions=None): + key = self._engine._extract_partition_key(context) + lattice = self._map.get(key) + if lattice is None: + lattice = self._route(key) + return self._engine.apply(lattice, context, dimensions=dimensions) +``` + +In `src/mountainash_rules/__init__.py`: add `AmbiguousPartitionError` to the `lattice` import line and to `__all__` (alphabetical — right after `Aggregate`). + +Note: the test helper above already passes `validate=`, so give `engine.index()` its final signature now — the parameters are accepted but not yet forwarded (Task 5 threads them into `LatticeIndex`): + +```python + def index( + self, + lattices: list[Lattice], + validate: bool = True, + max_witnesses: int = 1_000_000, + ) -> LatticeIndex: + """Build a partition-key routing index over pre-built lattices.""" + from mountainash_rules.engines.accumulator.lattice import LatticeIndex + return LatticeIndex(self, lattices, self._context_key_dims) +``` + +(Task 5 threads `validate`/`max_witnesses` into `LatticeIndex`.) + +- [ ] **Step 4: Run tests to verify they pass** + +Run: `hatch run test:test-target tests/accumulator/test_lattice.py::TestTernaryRouting tests/accumulator/test_lattice.py::TestIndexStructuralChecks -v` +Expected: all PASS. + +- [ ] **Step 5: Run the quick suite and commit** + +Run: `hatch run test:test-quick && hatch run ruff:check` +Expected: green — in particular `tests/accumulator/test_apply.py::TestApplyAutoWithPartitions` (fast path + `KeyError` message prefix preserved) and `tests/test_backend_purity.py`. + +```bash +git add src/mountainash_rules/engines/accumulator/lattice.py src/mountainash_rules/engines/accumulator/engine.py src/mountainash_rules/__init__.py tests/accumulator/test_lattice.py +git commit -m "feat: ternary + specificity partition routing via embedded meta-engine" +``` + +--- + +### Task 4: `apply_batch` routing through `_route` + +**Files:** +- Modify: `src/mountainash_rules/engines/accumulator/lattice.py` (`LatticeIndex.apply_batch`, the combo-loop only) +- Test: `tests/accumulator/test_lattice.py` + +**Interfaces:** +- Consumes: `_route` and `_map` (Task 3), `engine._normalize_partition_key` (Task 2). +- Produces: `apply_batch(contexts, **kwargs)` — per unique key combo: normalise (None → NOT_SET), dict hit first, `_route` on miss; combos routing to the same lattice batch together; unroutable combo raises the same errors as `apply`. Null combo values filter their partition slice with `is_null`. + +- [ ] **Step 1: Write the failing tests** + +Append to `tests/accumulator/test_lattice.py`: + +```python +class TestTernaryRoutingBatch: + def test_batch_mixes_specific_and_default(self): + engine, index = _index_for([("AU", "BROKER"), (UNKNOWN, UNKNOWN)]) + contexts = pl.DataFrame({ + "region": ["AU", "NZ", None], + "channel": ["BROKER", "DIRECT", None], + "product": ["GOLD", "GOLD", "GOLD"], + }) + result = index.apply_batch(contexts) + rows = relation(result.survivors).to_dict() + # every context produced at least one survivor row + assert set(rows["__lattice_ctx_id"]) == {0, 1, 2} + + def test_batch_unroutable_combo_raises(self): + engine, index = _index_for([("AU", "BROKER")]) + contexts = pl.DataFrame({ + "region": ["AU", "US"], + "channel": ["BROKER", "X"], + "product": ["GOLD", "GOLD"], + }) + with pytest.raises(KeyError, match="No lattice"): + index.apply_batch(contexts) + + def test_batch_ambiguous_combo_raises(self): + engine, index = _index_for( + [("AU", UNKNOWN), (UNKNOWN, "BROKER")], validate=False + ) + contexts = pl.DataFrame({ + "region": ["AU"], + "channel": ["BROKER"], + "product": ["GOLD"], + }) + with pytest.raises(AmbiguousPartitionError): + index.apply_batch(contexts) +``` + +`relation` is already imported at the top of `test_lattice.py` (`from mountainash.relations import relation`); add it if absent. + +- [ ] **Step 2: Run tests to verify they fail** + +Run: `hatch run test:test-target tests/accumulator/test_lattice.py::TestTernaryRoutingBatch -v` +Expected: `test_batch_mixes_specific_and_default` FAILS with `KeyError` (null combo has no exact dict entry); the unroutable test may already pass — that's fine, the mixed test is the driver. + +- [ ] **Step 3: Implement** + +In `LatticeIndex.apply_batch`, replace the combo loop body: + +```python + for i in range(n): + raw_key = tuple(combos[f][i] for f in key_fields) + key = self._engine._normalize_partition_key(raw_key) + lattice = self._map.get(key) + if lattice is None: + lattice = self._route(key) + part = rel + for f, v in zip(key_fields, raw_key): + part = part.filter( + ma.col(f).is_null() if v is None + else ma.col(f).eq(ma.lit(v)) + ) + engine = self._engine._filter_engine_for(lattice) + frames.append(relation(engine.evaluate_batch( + part.collect(), **kwargs + ).survivors)) +``` + +(Only the key resolution and the null-aware filter change; the id synthesis, concat, and `BatchRuleResult` wrap are untouched.) + +- [ ] **Step 4: Run tests to verify they pass** + +Run: `hatch run test:test-target tests/accumulator/test_lattice.py::TestTernaryRoutingBatch -v` +Expected: 3 PASS. Then `hatch run test:test-quick` — `tests/filter/test_batch_evaluation.py` index-batch tests must stay green. + +- [ ] **Step 5: Commit** + +```bash +git add src/mountainash_rules/engines/accumulator/lattice.py tests/accumulator/test_lattice.py +git commit -m "feat: route apply_batch combos through the shared ternary resolver" +``` + +--- + +### Task 5: Load-time ambiguity validation (witness matrix) + +**Files:** +- Modify: `src/mountainash_rules/engines/accumulator/lattice.py` (`LatticeIndex.__init__` signature + `_validate_ambiguity`) +- Modify: `src/mountainash_rules/engines/accumulator/engine.py` (`index` threads `validate`/`max_witnesses`) +- Test: `tests/accumulator/test_lattice.py` + +**Interfaces:** +- Consumes: `self._meta_engine.evaluate_batch`, `unknown_sentinel_for`, `not_set_sentinel_for` (all in place after Tasks 1–3). +- Produces: `LatticeIndex.__init__(engine, lattices, context_key_dims, validate=True, max_witnesses=1_000_000)`; private `_validate_ambiguity(max_witnesses)` raising `AmbiguousPartitionError` (with one witness context in the message) or `ValueError` on witness-cap overflow. `AccumulatorEngine.index(lattices, validate=True, max_witnesses=1_000_000)` passes both through. + +- [ ] **Step 1: Write the failing tests** + +Append to `tests/accumulator/test_lattice.py`: + +```python +class TestIndexValidation: + def test_crossing_pair_without_cover_raises_at_index(self): + engine = _routing_engine() + lattices = engine.build_all( + _routing_rules([("AU", UNKNOWN), (UNKNOWN, "BROKER")]) + ) + with pytest.raises(AmbiguousPartitionError) as exc: + engine.index(lattices) + # message carries a witness context + assert "AU" in str(exc.value) and "BROKER" in str(exc.value) + + def test_crossing_pair_with_cover_validates_and_routes(self): + engine = _routing_engine() + lattices = engine.build_all(_routing_rules([ + ("AU", UNKNOWN), (UNKNOWN, "BROKER"), ("AU", "BROKER"), + ])) + index = engine.index(lattices) # must NOT raise (false-positive guard) + result = index.apply( + RoutingContext(region="AU", channel="BROKER", product="GOLD") + ) + assert result.count >= 1 + + def test_validate_false_defers_to_runtime(self): + engine = _routing_engine() + lattices = engine.build_all( + _routing_rules([("AU", UNKNOWN), (UNKNOWN, "BROKER")]) + ) + index = engine.index(lattices, validate=False) # no raise here + with pytest.raises(AmbiguousPartitionError): + index.apply(RoutingContext(region="AU", channel="BROKER", product="GOLD")) + + def test_witness_cap_overflow_raises(self): + engine = _routing_engine() + lattices = engine.build_all( + _routing_rules([("AU", "BROKER"), ("NZ", "DIRECT")]) + ) + # 3 classes per dim (AU, NZ, OTHER) x (BROKER, DIRECT, OTHER) = 9 > 4 + with pytest.raises(ValueError, match="max_witnesses"): + engine.index(lattices, max_witnesses=4) + + def test_bool_key_dim_full_domain_validates(self): + metadata = DimensionsMetadata(dimensions=[ + Dimension( + dimension_name="flag", + match_strategy=MatchStrategy.EXACT, + data_type="bool", + role=DimensionRole.CONTEXT_KEY, + ), + Dimension(dimension_name="product", match_strategy=MatchStrategy.EXACT), + ]) + engine = AccumulatorEngine( + dimension_metadata=metadata, + aggregates=[Aggregate(column_name="margin")], + ) + rules = pl.DataFrame({ + "flag": [True, False, None], # null rule value = bool wildcard + "rule_name": ["r0", "r1", "r2"], + "product": ["GOLD"] * 3, + "margin": [1.0] * 3, + }) + index = engine.index(engine.build_all(rules)) # OTHER = None, no raise + result = index.apply({"product": "GOLD"}) # flag missing -> wildcard + assert result.count >= 1 +``` + +- [ ] **Step 2: Run tests to verify they fail** + +Run: `hatch run test:test-target tests/accumulator/test_lattice.py::TestIndexValidation -v` +Expected: `test_crossing_pair_without_cover_raises_at_index` and `test_witness_cap_overflow_raises` FAIL (no validation exists yet); the others may pass incidentally. + +- [ ] **Step 3: Implement** + +In `lattice.py`, extend `LatticeIndex.__init__`'s signature and tail: + +```python + def __init__( + self, + engine, + lattices: list["Lattice"], + context_key_dims, + validate: bool = True, + max_witnesses: int = 1_000_000, + ) -> None: +``` + +and after `self._meta_engine = ...`: + +```python + if validate and self._meta_engine is not None: + self._validate_ambiguity(max_witnesses) +``` + +Add the method (module needs `import itertools` at the top): + +```python + _WITNESS_CHUNK = 100_000 + + def _validate_ambiguity(self, max_witnesses: int) -> None: + """Exhaustive witness-matrix ambiguity check (spec §5). + + Per key dim the reachable context values collapse into finitely + many equivalence classes: each specific key value, plus OTHER — + represented by the typed NOT_SET sentinel (None for bool), which + matches no specific key and every wildcard. The cross-product of + classes is routed through the meta-engine in chunks; any witness + with >= 2 top-specificity survivors is a reachable runtime tie. + """ + classes: list[list] = [] + for i, d in enumerate(self._context_key_dims): + if d.data_type is DataType.BOOL: + wildcard, other = None, None + else: + wildcard = unknown_sentinel_for(d.data_type) + other = not_set_sentinel_for(d.data_type) + specifics = sorted( + {key[i] for key in self._map if key[i] != wildcard}, + key=repr, + ) + classes.append(specifics + [other]) + + total = 1 + for c in classes: + total *= len(c) + if total > max_witnesses: + raise ValueError( + f"Ambiguity validation needs {total} witness contexts, " + f"over max_witnesses={max_witnesses}; pass a higher " + f"max_witnesses, validate=False (accepting the runtime " + f"tie check), or restructure the key dimensions" + ) + + fields = [d.resolved_context_field for d in self._context_key_dims] + witnesses = itertools.product(*classes) + while True: + chunk = list(itertools.islice(witnesses, self._WITNESS_CHUNK)) + if not chunk: + return + contexts = relation({ + "__witness_id": list(range(len(chunk))), + **{ + f: [w[i] for w in chunk] + for i, f in enumerate(fields) + }, + }).collect() + survivors = relation( + self._meta_engine.evaluate_batch( + contexts, context_id_field="__witness_id" + ).survivors + ).to_dict() + best: dict[int, int] = {} + tied: dict[int, list[int]] = {} + for wid, spec, pidx in zip( + survivors["__witness_id"], + survivors["__specificity"], + survivors["__partition_idx"], + ): + if wid not in best or spec > best[wid]: + best[wid] = spec + tied[wid] = [pidx] + elif spec == best[wid]: + tied[wid].append(pidx) + for wid, parts in tied.items(): + if len(parts) > 1: + witness_ctx = dict(zip(fields, chunk[wid])) + tied_keys = [ + tuple( + self._lattices[p].partition_key[d.dimension_name] + for d in self._context_key_dims + ) + for p in parts + ] + raise AmbiguousPartitionError( + f"Partition suite is ambiguous: witness context " + f"{witness_ctx!r} ties partitions {tied_keys!r}" + ) +``` + +**Caveat for the implementer:** `evaluate_batch`'s `_check_reserved` rejects context frames containing reserved column prefixes. `__witness_id` is not in `_BATCH_RESERVED` and does not start with `__t_` or `__ctx_`, so it passes — verify this when running the tests; if it collides, rename to `witness_id` everywhere in this method. + +In `engine.py`, thread the parameters through `index`: + +```python + def index( + self, + lattices: list[Lattice], + validate: bool = True, + max_witnesses: int = 1_000_000, + ) -> LatticeIndex: + """Build a partition-key routing index over pre-built lattices. + + Args: + lattices: List of Lattice objects from build_all() or load(). + validate: Run the exhaustive load-time ambiguity check + (structural checks — empty/duplicate/NOT_SET keys — run + regardless). + max_witnesses: Ceiling on the validation matrix size; above + it index() raises ValueError rather than sampling. + """ + from mountainash_rules.engines.accumulator.lattice import LatticeIndex + return LatticeIndex( + self, lattices, self._context_key_dims, + validate=validate, max_witnesses=max_witnesses, + ) +``` + +- [ ] **Step 4: Run tests to verify they pass** + +Run: `hatch run test:test-target tests/accumulator/test_lattice.py::TestIndexValidation -v` +Expected: 5 PASS. **Important:** earlier routing tests built ambiguous suites with `validate=False` — re-run the whole file: `hatch run test:test-target tests/accumulator/test_lattice.py -v`. All green. + +- [ ] **Step 5: Run the full quick suite, lint, mypy, and commit** + +Run: `hatch run test:test-quick && hatch run ruff:check && hatch run mypy:check` +Expected: green. + +```bash +git add src/mountainash_rules/engines/accumulator/lattice.py src/mountainash_rules/engines/accumulator/engine.py tests/accumulator/test_lattice.py +git commit -m "feat: exhaustive load-time ambiguity validation for lattice indexes" +``` + +--- + +### Task 6: Persistence round-trip test + documentation + +**Files:** +- Test: `tests/accumulator/test_lattice.py` +- Modify: `CLAUDE.md` (Match Strategies table; accumulator engine section) + +**Interfaces:** +- Consumes: `Lattice.save/load` (unchanged), `engine.index` (Task 5). +- Produces: proof that wildcard sentinels round-trip through parquet + manifest and the loaded suite routes; updated repo docs. + +- [ ] **Step 1: Write the round-trip test** + +Append to `tests/accumulator/test_lattice.py` (the file already imports `Lattice`; `tmp_path` is a pytest builtin fixture): + +```python +class TestRoutingPersistence: + def test_saved_wildcard_suite_routes_after_load(self, tmp_path): + engine = _routing_engine() + built = engine.build_all( + _routing_rules([("AU", "BROKER"), (UNKNOWN, UNKNOWN)]) + ) + loaded = [ + Lattice.load(lattice.save(tmp_path / f"part{i}")) + for i, lattice in enumerate(built) + ] + index = engine.index(loaded) + result = index.apply( + RoutingContext(region="NZ", channel="DIRECT", product="GOLD") + ) + assert result.count >= 1 # sentinel key survived the round-trip +``` + +- [ ] **Step 2: Run it** + +Run: `hatch run test:test-target tests/accumulator/test_lattice.py::TestRoutingPersistence -v` +Expected: PASS (no production change in this task — this is the spec §8 persistence guarantee). If it fails, stop and diagnose (likely sentinel mangling in manifest YAML); do not weaken the test. + +- [ ] **Step 3: Update CLAUDE.md** + +In the Match Strategies table, after the `exact` / `not_equal` row, add: + +```markdown +| `exact_key` | scalar | rule-side wildcard only (UNKNOWN → 0; context sentinel vs specific → −1); powers partition routing | +``` + +In the "Accumulator engine" section, replace the sentence beginning "`DimensionRole.CONTEXT_KEY` dimensions partition the rule space" with: + +```markdown +- `DimensionRole.CONTEXT_KEY` dimensions partition the rule space; `build_all` + `index(lattices)` → `LatticeIndex` routes contexts (single or batch) to the right lattice using ternary + specificity semantics via an embedded meta-engine (`EXACT_KEY` dims): an all-wildcard key is the default/overflow partition, exact hits keep an O(1) dict fast path, ties raise `AmbiguousPartitionError` (a `KeyError` subclass, exported from the package root). `index(lattices, validate=True, max_witnesses=1_000_000)` runs an exhaustive witness-matrix ambiguity check at load time; empty/duplicate/NOT_SET-bearing keys are always rejected. See `docs/superpowers/specs/2026-07-19-ternary-partition-routing-design.md`. +``` + +- [ ] **Step 4: Final full verification** + +Run: `hatch run test:test && hatch run ruff:check && hatch run mypy:check` +Expected: full suite + coverage green. + +- [ ] **Step 5: Commit** + +```bash +git add tests/accumulator/test_lattice.py CLAUDE.md +git commit -m "test: routing persistence round-trip; docs: ternary partition routing" +``` + +--- + +## Post-implementation + +Per the standard delivery flow: adversarial post-implementation review (`/codex:rescue` on the branch diff), then PR targeting `develop` (`gh pr create --base develop`). Do not push to `develop` or `main` directly. diff --git a/docs/superpowers/specs/2026-07-19-ternary-partition-routing-design.md b/docs/superpowers/specs/2026-07-19-ternary-partition-routing-design.md index 2acc8a1..a5612fa 100644 --- a/docs/superpowers/specs/2026-07-19-ternary-partition-routing-design.md +++ b/docs/superpowers/specs/2026-07-19-ternary-partition-routing-design.md @@ -1,6 +1,7 @@ # Ternary Partition Routing — Design Spec -> **Status:** APPROVED (design walkthrough 2026-07-19) +> **Status:** APPROVED (design walkthrough 2026-07-19; amended same day after +> Codex adversarial review — see §9 Review adjudication) > **Source:** `mountainash-central/01.principles/mountainash-rules/h.backlog/ternary-partition-routing.md` > **Repo:** `mountainash-rules`, branch `feature/ternary-partition-routing` (off `develop` @ `496f84a`) @@ -47,15 +48,35 @@ Per CONTEXT_KEY dimension, context value vs partition key value: - **All-wildcard key = default/overflow partition**: survives every context at specificity 0, wins only when nothing more specific survives. - A context value equal to the dim's UNKNOWN sentinel is treated as unknown: - it matches wildcard keys (0), never specific keys (−1) — consistent with - the expression layer's in-band sentinel handling. + it matches wildcard keys (0), never specific keys (−1). +- **These semantics are asymmetric** — only the *rule side* (partition key) + has a wildcard; a context-side sentinel is a non-match against any specific + key. The stock EXACT compile does **not** deliver this: `_compile_exact` + uses `t_eq` with both UNKNOWN and NOT_SET in the sentinel set on *both* + sides, so a NOT_SET-filled context field would score 0 against every + specific key — every partition would survive missing-field contexts and + tie spuriously. Routing therefore uses a dedicated strategy, §2's + `EXACT_KEY`. +- **Bool key dims:** the wildcard is `null` on the key side (bools have no + in-band sentinel; `_compile_bool_ternary` already uses null as don't-care). + A context-side `None` is wildcard-only, mirroring NOT_SET. **Missing context key fields.** `_extract_partition_key` fills a missing context field with the typed NOT_SET sentinel -(`not_set_sentinel_for(d.data_type)`) instead of raising `KeyError` — -mirroring `core/context.py`'s fill for constraint dims. A NOT_SET value -matches wildcard keys only, so a context with no `region` routes to the -default partition if one exists, else raises the no-partition error. +(`not_set_sentinel_for(d.data_type)`; `None` for bool) instead of raising +`KeyError` — mirroring `core/context.py`'s fill for constraint dims. An +**explicitly-null** context value (`region: None`, backend null/NaN) is +treated identically to an absent field. A NOT_SET value matches wildcard +keys only, so a context with no `region` routes to the default partition if +one exists, else raises the no-partition error. + +**Partition keys may not contain NOT_SET sentinels.** `index()` rejects any +lattice whose key contains a NOT_SET sentinel (`ValueError`, always-on — +independent of the `validate` flag). Without this, a NOT_SET-filled context +tuple could exact-dict-hit such a key and route a missing-field context to a +specific partition, violating the semantics above. NOT_SET keys cannot arise +from `build_all` (it groups on rule values, and NOT_SET is a context-side +sentinel); this guards hand-assembled suites. **Outcomes of routing one context:** @@ -81,10 +102,31 @@ no-survivor case deliberately stays a plain `KeyError` (same type as today). means wildcard in-band — no encoding step); plus a `__partition_idx` integer payload column mapping a surviving row back to its `Lattice`. - **Meta-engine** — an internal `ExpressionRulesEngine` over the meta-table. - Each CONTEXT_KEY dim is recast as an EXACT-strategy CONSTRAINT dimension: - same `dimension_name` and `data_type`, `rule_field` = the meta-table - column, `context_field` = the original dim's `resolved_context_field`. - Hit policy COLLECT (selection is done by the resolver, not a policy). + Each CONTEXT_KEY dim is recast as an **`EXACT_KEY`**-strategy CONSTRAINT + dimension: same `dimension_name` and `data_type`, `rule_field` = the + meta-table column, `context_field` = the original dim's + `resolved_context_field`. Hit policy COLLECT (selection is done by the + resolver, not a policy). + +**`MatchStrategy.EXACT_KEY`** — a new strategy added via the established +extension path (enum member in `core/constants.py`, validation in +`core/dimension.py`, `_compile_exact_key` in `core/compiler.py`, test class +in `tests/core/test_compiler.py`). Semantics: **rule-side wildcard only** — + +``` +rule value == UNKNOWN sentinel → 0 +rule value == context value → 1 +otherwise → −1 +``` + +i.e. `t_col(rule_field, unknown={unknown_sentinel_for(dt)}).t_eq(col(ctx))` +— the rule side recognises only the UNKNOWN sentinel (not NOT_SET, which is +forbidden in keys anyway), and the context side is a plain column, so +context-side sentinels compare as ordinary non-matching values → −1. Bool +variant mirrors `_compile_bool_ternary` but tests **rule-side null only**. +This is the same one-engine architecture — `EXACT_KEY` is a first-class +strategy any consumer may use, compiled and evaluated by the single +constraint pipeline; routing just happens to be its first consumer. Routing = `meta_engine.evaluate(context)`; the survivor frame's `__specificity` column drives winner / tie / miss detection. There is **one** @@ -106,11 +148,17 @@ lattice = self._route(context) # meta-engine evaluate return engine.apply(lattice, context, dimensions=...) ``` -The exact fast path is semantics-preserving: an exact hit matches every key -dim at 1, which is provably maximal specificity, and duplicate exact keys -cannot coexist (validation §5 / dict construction). A wildcard-free index -never reaches `_route`, so today's behaviour is byte-identical for existing -users. +The exact fast path is semantics-preserving, including for keys containing +UNKNOWN wildcards. Proof sketch: a dict hit means the context tuple equals +key `K` exactly (keys cannot contain NOT_SET — §1 — so NOT_SET-filled +contexts never false-hit). Partition `K` scores 1 on each of its specific +dims and 0 on its wildcard dims → specificity = |specific dims of K|. Any +competitor must be wildcard on every dim where `K` is wildcard (the context +carries the UNKNOWN sentinel there, which kills specific keys at −1), and +scores ≤ 1 elsewhere — so its specificity ≤ `K`'s, with equality only for an +identical key, which the dict (plus §5 duplicate detection) rules out. `K` +is the unique top-specificity survivor. A wildcard-free index never reaches +`_route`, so today's behaviour is byte-identical for existing users. `_route(context)` — the shared resolver: evaluate on the meta-engine; apply the outcome table from §1. @@ -131,6 +179,18 @@ The no-key-dims early path (single flat lattice) is unchanged. `engine.index(lattices, validate=True)` — **on by default**. +**Structural checks — always on, independent of `validate` (O(n), no +matrix):** run over the input *list* before dict construction, because the +dict silently collapses duplicates and can never see them afterwards: + +- `index([])` → `ValueError` (no lattices; also removes any ambiguity about + the meta-table's schema seed). +- Duplicate partition keys across the input list → `ValueError` naming the + key (previously one lattice was silently discarded by the dict). +- Any key containing a NOT_SET sentinel → `ValueError` (§1). + +**Ambiguity check (`validate=True`):** + Naïve pairwise overlap detection would false-positive the legitimate "more specific partition covers the crossing" pattern: `(AU, *)` vs `(*, BROKER)` is genuinely ambiguous **unless** `(AU, BROKER)` also exists, @@ -139,8 +199,14 @@ The check is therefore exact, built on a finite abstraction: 1. Per key dim, reachable context values fall into finitely many equivalence classes: each specific value appearing in any partition key, - plus one OTHER representative (a fresh value matching no specific key — - derived per `data_type`). Routing behaviour depends only on the class. + plus one OTHER representative — **the dim's typed NOT_SET sentinel + (`None` for bool)**. This is always constructible (no "fresh value" + derivation, no finite-domain problem for bool) and provably matches no + specific key: NOT_SET is forbidden in keys (§1) and under `EXACT_KEY` a + context-side sentinel scores −1 against every specific value and 0 + against wildcards — exactly the OTHER class's behaviour. It also *is* a + reachable context (a missing field), so every witness is a realisable + input. Routing behaviour depends only on the class. 2. The cross-product of classes forms a **witness-context matrix** that provably exercises every distinguishable routing case. 3. Route the whole matrix through the meta-engine as one `evaluate_batch` — @@ -152,12 +218,20 @@ The check is therefore exact, built on a finite abstraction: Properties: -- **Duplicate keys** are the degenerate tie — caught by the same mechanism - (dict construction also collapses them; validation makes it loud). -- Matrix size = `∏(distinct specific values per dim + 1)`; vectorised in one - batch, trivial at realistic sizes. `validate=False` is the opt-out for - pathological indexes; the runtime tie check in `_route` remains as a - backstop reachable only via that opt-out. +- **Duplicate keys** are caught by the structural pre-check above, *not* by + the matrix — dict construction collapses them before the meta-table + exists, so the matrix could never see a duplicate. +- Matrix size = `∏(distinct specific values per dim + 1)` — bounded by + `(P+1)^D` for `P` partitions over `D` key dims, so it can explode for + many-dimensional suites. Two guards: (a) the matrix is **evaluated in + fixed-size chunks** (default 100 000 witnesses per `evaluate_batch`), so + memory stays bounded regardless of total size; (b) a **witness-count cap** + (default 1 000 000, exposed as `max_witnesses` on `index()`) above which + `index()` raises `ValueError` telling the caller to pass `validate=False` + (accepting the runtime backstop) or restructure the key dims. No silent + sampling — validation is exact or explicitly declined. +- `validate=False` is the opt-out; the runtime tie check in `_route` remains + as a backstop reachable only via that opt-out. - **Service inherits load-time detection with no change:** `_load_partitioned` calls `engine.index(lattices)` inside its per-slug try/except, so an ambiguous suite is skipped with a warning at startup. @@ -175,8 +249,9 @@ instead of dead weight. The footgun is resolved by routing, not building. |---|---| | `LatticeIndex.apply` | wildcard + specificity routing; `KeyError` message now lists served partitions; may raise `AmbiguousPartitionError` | | `LatticeIndex.apply_batch` | same routing per unique combo | -| `AccumulatorEngine.index` | gains `validate: bool = True` | -| `AccumulatorEngine._extract_partition_key` | missing context field → typed NOT_SET fill (was `KeyError`) | +| `AccumulatorEngine.index` | gains `validate: bool = True`, `max_witnesses: int = 1_000_000`; raises `ValueError` on empty list, duplicate keys, NOT_SET-bearing keys, or witness-cap overflow | +| `AccumulatorEngine._extract_partition_key` | missing **or explicitly-null** context field → typed NOT_SET fill (`None` for bool) (was `KeyError`) | +| `MatchStrategy.EXACT_KEY` | new enum member — rule-side-wildcard-only exact match (§2); first-class strategy, usable by any consumer | | `AmbiguousPartitionError` | new, subclasses `KeyError`, exported from package root `__all__` | | `apply_auto` | inherits all of the above (delegates to `index().apply`) | @@ -203,18 +278,49 @@ In `tests/accumulator/test_lattice.py` (new `TestTernaryRouting` / specific + default partitions in one batch with correct per-context groups; a batch containing an unroutable combo raises. +**Compiler (`tests/core/test_compiler.py`, new `TestExactKey` class):** +- rule UNKNOWN sentinel → 0 against any context value; +- rule specific vs equal context → 1; vs different → −1; +- rule specific vs context NOT_SET sentinel → **−1** (the asymmetry that + distinguishes `EXACT_KEY` from `EXACT`); +- rule specific vs context UNKNOWN sentinel → −1; +- bool: rule null → 0; rule specific vs context null → −1. + **Validation (`index()`):** - crossing pair, no cover → `AmbiguousPartitionError` at `index()`, message carries a witness context; - crossing pair + covering `(AU, BROKER)` partition → validates clean AND routes correctly at runtime (the false-positive guard); -- duplicate partition keys → caught; +- duplicate partition keys in the input list → `ValueError` (even with + `validate=False` — structural check); +- key containing a NOT_SET sentinel → `ValueError`; +- `index([])` → `ValueError`; +- witness count over `max_witnesses` → `ValueError` naming the opt-outs; +- bool key dim with both `True` and `False` as specific keys + a wildcard → + validates clean; context `{flag: None}` routes to the wildcard (OTHER + representative exists for finite domains); - `validate=False` defers the crossing pair to the runtime error. **Persistence:** `save` → `load` a wildcard-keyed partition suite → `index()` → default routing works (sentinel round-trips through parquet + manifest). +## 9. Review adjudication (Codex adversarial review, 2026-07-19) + +Nine findings; eight accepted and folded into the sections above: + +| Finding | Disposition | +|---|---| +| NOT_SET partition key can false-hit the fast path (critical) | Accepted — NOT_SET forbidden in keys, always-on check (§1, §5) | +| Duplicate keys collapse in the dict before validation | Accepted — structural pre-check over the input list (§5) | +| Wildcard routing via stock EXACT is unspecified | Accepted & confirmed against code (`_compile_exact` t_eq scores context NOT_SET as 0, not −1) — new `EXACT_KEY` strategy (§1, §2) | +| Witness matrix unbounded | Accepted — chunked evaluation + `max_witnesses` cap (§5) | +| OTHER representative undefined for bool | Accepted — OTHER = typed NOT_SET sentinel / `None` (§5) | +| Null context values unspecified | Accepted — explicit null ≡ missing (§1) | +| `index([])` undefined | Accepted — `ValueError` (§5) | +| Exact-hit specificity claim too broad | Accepted — proof rewritten to cover UNKNOWN-bearing keys (§3) | +| `AmbiguousPartitionError(KeyError)` → HTTP 422 misclassifies a config defect; startup skip hides it | **Rejected — deliberate tradeoff.** With `validate=True` default, ambiguity surfaces at `index()`/startup inside the service's existing per-slug skip-and-warn envelope (its standing failure mode for any bad suite). Runtime ambiguity is reachable only via explicit `validate=False`, at which point 422-on-tie is the same contract as today's partition miss. Zero-service-change wins; revisit if the service ever grows a config-health endpoint. | + ## Non-goals - Range/regex/other match strategies at the partition level — CONTEXT_KEY From 12a43607cb21290eef33a187a3ef11ed0b6b637a Mon Sep 17 00:00:00 2001 From: Nathaniel Ramm Date: Sun, 19 Jul 2026 18:48:44 +1000 Subject: [PATCH 02/11] =?UTF-8?q?feat:=20add=20MatchStrategy.EXACT=5FKEY?= =?UTF-8?q?=20=E2=80=94=20rule-side-wildcard-only=20exact=20match?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- src/mountainash_rules/core/compiler.py | 24 ++++++++ src/mountainash_rules/core/constants.py | 1 + tests/core/test_compiler.py | 79 ++++++++++++++++++++++++- 3 files changed, 103 insertions(+), 1 deletion(-) diff --git a/src/mountainash_rules/core/compiler.py b/src/mountainash_rules/core/compiler.py index 1ec07c8..a7d5e6a 100644 --- a/src/mountainash_rules/core/compiler.py +++ b/src/mountainash_rules/core/compiler.py @@ -12,6 +12,7 @@ DataType, MatchStrategy, sentinels_for, + unknown_sentinel_for, ) from mountainash_rules.core.dimension import Dimension, DimensionsMetadata @@ -35,6 +36,8 @@ def compile_dimension(self, dim: Dimension) -> BaseExpressionAPI: match dim.match_strategy: case MatchStrategy.EXACT: return self._compile_exact(dim) + case MatchStrategy.EXACT_KEY: + return self._compile_exact_key(dim) case MatchStrategy.RANGE: return self._compile_range(dim) case MatchStrategy.REGEX: @@ -68,6 +71,27 @@ def _compile_exact(self, dim: Dimension) -> BaseExpressionAPI: ctx_col = ma.t_col(CTX_PREFIX + dim.dimension_name, unknown=sentinels) return rule_col.t_eq(ctx_col) + def _compile_exact_key(self, dim: Dimension) -> BaseExpressionAPI: + """Rule-side-wildcard-only exact match (partition-key routing). + + Only the rule side has a wildcard (UNKNOWN sentinel; null for + bool). A context-side sentinel is an ordinary non-matching value: + specific keys must never match an unknown/unset context. + """ + rule_col = ma.col(dim.resolved_rule_field) + ctx_col = ma.col(CTX_PREFIX + dim.dimension_name) + if dim.data_type is DataType.BOOL: + wildcard = rule_col.is_null() + else: + wildcard = rule_col.__eq__( + ma.lit(unknown_sentinel_for(dim.data_type)) + ) + return ( + ma.when(wildcard).then(0) + .when(rule_col.__eq__(ctx_col)).then(1) + .otherwise(-1) + ) + def _compile_not_equal(self, dim: Dimension) -> BaseExpressionAPI: if dim.data_type is DataType.BOOL: return self._compile_bool_ternary(dim, "__ne__") diff --git a/src/mountainash_rules/core/constants.py b/src/mountainash_rules/core/constants.py index 8c857fb..05cabf4 100644 --- a/src/mountainash_rules/core/constants.py +++ b/src/mountainash_rules/core/constants.py @@ -8,6 +8,7 @@ class MatchStrategy(StrEnum): """How a dimension matches context values against rule values.""" EXACT = "exact" + EXACT_KEY = "exact_key" NOT_EQUAL = "not_equal" RANGE = "range" GREATER_THAN = "greater_than" diff --git a/tests/core/test_compiler.py b/tests/core/test_compiler.py index 0fe6b29..d07ec70 100644 --- a/tests/core/test_compiler.py +++ b/tests/core/test_compiler.py @@ -7,7 +7,7 @@ import mountainash.expressions as ma from mountainash_rules.core.compiler import DimensionCompiler -from mountainash_rules.core.constants import CTX_PREFIX, UNKNOWN, UNKNOWN_NUMERIC, MatchStrategy +from mountainash_rules.core.constants import CTX_PREFIX, NOT_SET, UNKNOWN, UNKNOWN_NUMERIC, MatchStrategy from mountainash_rules.core.dimension import Dimension from tests.conftest import ( ALL_BACKENDS, @@ -647,3 +647,80 @@ def test_set_strategy_compiles_on_backend(self, compiler, backend_name, strategy df = build_backend_df(backend_name, self._SAMPLE_DATA) compiled = expr.compile(df, booleanizer=None) assert compiled is not None + + +class TestExactKeyCompilation: + """EXACT_KEY: rule-side wildcard only — context sentinels are non-matches.""" + + def test_rule_unknown_is_wildcard(self, compiler): + dim = Dimension( + dimension_name="region", + match_strategy=MatchStrategy.EXACT_KEY, + data_type="str", + ) + expr = compiler.compile_dimension(dim) + df = pl.DataFrame({ + "region": ["AU", UNKNOWN, "UK"], + f"{CTX_PREFIX}region": ["AU", "AU", "AU"], + }) + result = df.with_columns(expr.name.alias("__t_region").compile(df, booleanizer=None)) + assert result["__t_region"].to_list() == [1, 0, -1] + + def test_context_not_set_never_matches_specific(self, compiler): + # The asymmetry that distinguishes EXACT_KEY from EXACT: + # a NOT_SET context is -1 against specific keys (EXACT gives 0). + dim = Dimension( + dimension_name="region", + match_strategy=MatchStrategy.EXACT_KEY, + data_type="str", + ) + expr = compiler.compile_dimension(dim) + df = pl.DataFrame({ + "region": ["AU", UNKNOWN], + f"{CTX_PREFIX}region": [NOT_SET, NOT_SET], + }) + result = df.with_columns(expr.name.alias("__t_region").compile(df, booleanizer=None)) + assert result["__t_region"].to_list() == [-1, 0] + + def test_context_unknown_never_matches_specific(self, compiler): + dim = Dimension( + dimension_name="region", + match_strategy=MatchStrategy.EXACT_KEY, + data_type="str", + ) + expr = compiler.compile_dimension(dim) + df = pl.DataFrame({ + "region": ["AU", UNKNOWN], + f"{CTX_PREFIX}region": [UNKNOWN, UNKNOWN], + }) + result = df.with_columns(expr.name.alias("__t_region").compile(df, booleanizer=None)) + assert result["__t_region"].to_list() == [-1, 0] + + def test_numeric_sentinels(self, compiler): + dim = Dimension( + dimension_name="product_id", + match_strategy=MatchStrategy.EXACT_KEY, + data_type="int", + ) + expr = compiler.compile_dimension(dim) + df = pl.DataFrame({ + "product_id": [1, UNKNOWN_NUMERIC, 2], + f"{CTX_PREFIX}product_id": [1, 1, 1], + }) + result = df.with_columns(expr.name.alias("__t_product_id").compile(df, booleanizer=None)) + assert result["__t_product_id"].to_list() == [1, 0, -1] + + def test_bool_rule_null_is_wildcard_context_null_is_not(self, compiler): + dim = Dimension( + dimension_name="flag", + match_strategy=MatchStrategy.EXACT_KEY, + data_type="bool", + ) + expr = compiler.compile_dimension(dim) + df = pl.DataFrame({ + "flag": [True, None, True, None], + f"{CTX_PREFIX}flag": [True, True, None, None], + }) + result = df.with_columns(expr.name.alias("__t_flag").compile(df, booleanizer=None)) + # rule null -> 0 regardless of context; context null vs specific -> -1 + assert result["__t_flag"].to_list() == [1, 0, -1, 0] From d840accc3d4f80b4a61a87ae8ebc3e727bed230d Mon Sep 17 00:00:00 2001 From: Nathaniel Ramm Date: Sun, 19 Jul 2026 18:54:54 +1000 Subject: [PATCH 03/11] feat: NOT_SET fill for missing/null partition key fields Co-Authored-By: Claude Opus 4.8 (1M context) --- .../engines/accumulator/engine.py | 21 ++++++++-- tests/accumulator/test_apply.py | 40 ++++++++++++++++++- 2 files changed, 57 insertions(+), 4 deletions(-) diff --git a/src/mountainash_rules/engines/accumulator/engine.py b/src/mountainash_rules/engines/accumulator/engine.py index eb8a9e7..db7f52b 100644 --- a/src/mountainash_rules/engines/accumulator/engine.py +++ b/src/mountainash_rules/engines/accumulator/engine.py @@ -17,9 +17,11 @@ from mountainash_rules.engines.accumulator.result import AccumulatorResult from mountainash_rules.engines.accumulator.aggregate import Aggregate from mountainash_rules.core.constants import ( + DataType, DimensionRole, HitPolicy, MatchStrategy, + not_set_sentinel_for, unknown_sentinel_for, ) from mountainash_rules.core.dimension import Dimension, DimensionsMetadata @@ -546,7 +548,12 @@ def index(self, lattices: list[Lattice]) -> LatticeIndex: return LatticeIndex(self, lattices, self._context_key_dims) def _extract_partition_key(self, context: t.Any) -> tuple: - """Extract the partition key tuple from a context object.""" + """Extract the partition key tuple from a context object. + + Missing or explicitly-null key fields become the typed NOT_SET + sentinel (None for bool) so the context can still route — a + NOT_SET value matches wildcard partitions only. + """ if isinstance(context, BaseModel): raw = context.model_dump() elif isinstance(context, dict): @@ -555,9 +562,17 @@ def _extract_partition_key(self, context: t.Any) -> tuple: raise TypeError( f"Context must be a BaseModel or dict, got {type(context).__name__}" ) - return tuple( - raw[d.resolved_context_field] + return self._normalize_partition_key(tuple( + raw.get(d.resolved_context_field) for d in self._context_key_dims + )) + + def _normalize_partition_key(self, key: tuple) -> tuple: + """Map None key values to the typed NOT_SET sentinel (None for bool).""" + return tuple( + v if v is not None or d.data_type is DataType.BOOL + else not_set_sentinel_for(d.data_type) + for v, d in zip(key, self._context_key_dims) ) def apply_auto( diff --git a/tests/accumulator/test_apply.py b/tests/accumulator/test_apply.py index b027a0a..da1dc1c 100644 --- a/tests/accumulator/test_apply.py +++ b/tests/accumulator/test_apply.py @@ -8,7 +8,7 @@ from mountainash_rules.engines.accumulator.engine import AccumulatorEngine from mountainash_rules.engines.accumulator.result import AccumulatorResult from mountainash_rules.engines.accumulator.aggregate import Aggregate -from mountainash_rules.core.constants import UNKNOWN, UNKNOWN_NUMERIC, MatchStrategy, DimensionRole +from mountainash_rules.core.constants import UNKNOWN, UNKNOWN_NUMERIC, NOT_SET_NUMERIC, MatchStrategy, DimensionRole from mountainash_rules.core.dimension import Dimension, DimensionsMetadata @@ -154,3 +154,41 @@ def test_apply_auto_missing_key_raises(self): with pytest.raises(KeyError, match="No lattice"): engine.apply_auto(lattices, PartitionedContext(product_id=99, channel="X")) + + +class TestExtractPartitionKey: + def _engine(self): + metadata = DimensionsMetadata(dimensions=[ + Dimension( + dimension_name="product_id", + match_strategy=MatchStrategy.EXACT, + data_type=int, + role=DimensionRole.CONTEXT_KEY, + ), + Dimension(dimension_name="channel", match_strategy=MatchStrategy.EXACT), + ]) + return AccumulatorEngine( + dimension_metadata=metadata, + aggregates=[Aggregate(column_name="margin")], + ) + + def test_missing_key_field_fills_not_set(self): + key = self._engine()._extract_partition_key({"channel": "BROKER"}) + assert key == (NOT_SET_NUMERIC,) + + def test_explicit_none_fills_not_set(self): + key = self._engine()._extract_partition_key( + {"product_id": None, "channel": "BROKER"} + ) + assert key == (NOT_SET_NUMERIC,) + + def test_present_value_passes_through(self): + key = self._engine()._extract_partition_key( + {"product_id": 7, "channel": "BROKER"} + ) + assert key == (7,) + + def test_normalize_partition_key(self): + engine = self._engine() + assert engine._normalize_partition_key((None,)) == (NOT_SET_NUMERIC,) + assert engine._normalize_partition_key((7,)) == (7,) From e067ff5215ff6441c47c9147b9c9c24be3be87d2 Mon Sep 17 00:00:00 2001 From: Nathaniel Ramm Date: Sun, 19 Jul 2026 19:01:27 +1000 Subject: [PATCH 04/11] feat: ternary + specificity partition routing via embedded meta-engine --- src/mountainash_rules/__init__.py | 7 +- .../engines/accumulator/engine.py | 7 +- .../engines/accumulator/lattice.py | 120 ++++++++++++++- tests/accumulator/test_lattice.py | 141 ++++++++++++++++++ 4 files changed, 267 insertions(+), 8 deletions(-) diff --git a/src/mountainash_rules/__init__.py b/src/mountainash_rules/__init__.py index 0133b37..4d5cc2c 100644 --- a/src/mountainash_rules/__init__.py +++ b/src/mountainash_rules/__init__.py @@ -26,7 +26,11 @@ from mountainash_rules.core.batch_result import BatchRuleResult from mountainash_rules.engines.filter.engine import ExpressionRulesEngine from mountainash_rules.core.hit_policy import HitPolicyViolationError, SelectionInfo -from mountainash_rules.engines.accumulator.lattice import Lattice, LatticeIndex +from mountainash_rules.engines.accumulator.lattice import ( + AmbiguousPartitionError, + Lattice, + LatticeIndex, +) from mountainash_rules.core.result import ExplainResult, RuleResult __all__ = ( @@ -34,6 +38,7 @@ "AccumulatorEngine", "AccumulatorResult", "Aggregate", + "AmbiguousPartitionError", "BatchRuleResult", "DataType", "DimensionCompiler", diff --git a/src/mountainash_rules/engines/accumulator/engine.py b/src/mountainash_rules/engines/accumulator/engine.py index db7f52b..d392177 100644 --- a/src/mountainash_rules/engines/accumulator/engine.py +++ b/src/mountainash_rules/engines/accumulator/engine.py @@ -542,7 +542,12 @@ def _filter_engine_for(self, lattice: Lattice) -> ExpressionRulesEngine: self._apply_engines[lattice] = engine return engine - def index(self, lattices: list[Lattice]) -> LatticeIndex: + def index( + self, + lattices: list[Lattice], + validate: bool = True, + max_witnesses: int = 1_000_000, + ) -> LatticeIndex: """Build a partition-key routing index over pre-built lattices.""" from mountainash_rules.engines.accumulator.lattice import LatticeIndex return LatticeIndex(self, lattices, self._context_key_dims) diff --git a/src/mountainash_rules/engines/accumulator/lattice.py b/src/mountainash_rules/engines/accumulator/lattice.py index 88d2662..29f9ad4 100644 --- a/src/mountainash_rules/engines/accumulator/lattice.py +++ b/src/mountainash_rules/engines/accumulator/lattice.py @@ -5,7 +5,14 @@ from mountainash.relations import relation from mountainash_rules.engines.accumulator.aggregate import Aggregate -from mountainash_rules.core.dimension import DimensionsMetadata +from mountainash_rules.core.constants import ( + DataType, + HitPolicy, + MatchStrategy, + not_set_sentinel_for, +) +from mountainash_rules.core.dimension import Dimension, DimensionsMetadata +from mountainash_rules.engines.filter.engine import ExpressionRulesEngine class Lattice: @@ -100,28 +107,129 @@ def load(cls, dir_path: "str | pathlib.Path") -> "Lattice": ) +class AmbiguousPartitionError(KeyError): + """Two or more partitions tie at top specificity for a context. + + Subclasses KeyError so existing partition-miss handlers keep working; + reachable at runtime only when index() validation was opted out. + """ + + class LatticeIndex: - """Partition-key routing over a set of built lattices, built once.""" + """Partition-key routing over a set of built lattices, built once. + + Routing uses the same ternary + specificity semantics as the + constraint layer, evaluated by an embedded ExpressionRulesEngine over + a meta rules-table (one row per partition). Exact key hits keep an + O(1) dict fast path. + """ def __init__(self, engine, lattices: list["Lattice"], context_key_dims) -> None: self._engine = engine self._context_key_dims = list(context_key_dims) + self._lattices = list(lattices) + if not self._lattices: + raise ValueError("index() requires at least one lattice") + self._map: dict[tuple, Lattice] = {} - for lattice in lattices: + for lattice in self._lattices: if lattice.partition_key is not None: key = tuple( lattice.partition_key[d.dimension_name] for d in self._context_key_dims ) + elif self._context_key_dims: + raise ValueError( + "Lattice without a partition_key cannot be indexed " + "alongside CONTEXT_KEY dimensions" + ) else: key = () + for d, v in zip(self._context_key_dims, key): + if ( + d.data_type is not DataType.BOOL + and v == not_set_sentinel_for(d.data_type) + ): + raise ValueError( + f"Partition key {key!r} contains the NOT_SET " + f"sentinel for dimension '{d.dimension_name}'; " + f"NOT_SET is a context-side sentinel and would " + f"collide with missing-field routing" + ) + if key in self._map: + raise ValueError(f"Duplicate partition key {key!r}") self._map[key] = lattice + self._meta_engine = ( + self._build_meta_engine() if self._context_key_dims else None + ) + + def _build_meta_engine(self) -> "ExpressionRulesEngine": + """One meta-rule row per partition; EXACT_KEY per key dim.""" + columns: dict[str, list] = { + d.dimension_name: [] for d in self._context_key_dims + } + columns["__partition_idx"] = [] + for idx, lattice in enumerate(self._lattices): + for d in self._context_key_dims: + columns[d.dimension_name].append( + lattice.partition_key[d.dimension_name] + ) + columns["__partition_idx"].append(idx) + meta_metadata = DimensionsMetadata( + dimensions=[ + Dimension( + dimension_name=d.dimension_name, + context_field=d.resolved_context_field, + match_strategy=MatchStrategy.EXACT_KEY, + data_type=d.data_type, + ) + for d in self._context_key_dims + ], + hit_policy=HitPolicy.COLLECT, + ) + return ExpressionRulesEngine( + rules=relation(columns).collect(), + dimension_metadata=meta_metadata, + ) + + def _route(self, key: tuple) -> "Lattice": + """Ternary + specificity routing for a normalised key tuple.""" + ctx = { + d.resolved_context_field: key[i] + for i, d in enumerate(self._context_key_dims) + } + rows = relation(self._meta_engine.evaluate(ctx).survivors).to_dict() + idxs = rows["__partition_idx"] + if not idxs: + raise KeyError( + f"No lattice for partition key {key!r}; served partition " + f"keys: {sorted(self._map)!r}" + ) + top = rows["__specificity"][0] # survivors are rank-sorted + tied = [ + i for i, s in zip(idxs, rows["__specificity"]) if s == top + ] + if len(tied) > 1: + tied_keys = [ + tuple( + self._lattices[i].partition_key[d.dimension_name] + for d in self._context_key_dims + ) + for i in tied + ] + raise AmbiguousPartitionError( + f"Context key {key!r} ties {len(tied)} partitions at " + f"specificity {top}: {tied_keys!r}" + ) + return self._lattices[tied[0]] + def apply(self, context, dimensions=None): key = self._engine._extract_partition_key(context) - if key not in self._map: - raise KeyError(f"No lattice for partition key {key!r}") - return self._engine.apply(self._map[key], context, dimensions=dimensions) + lattice = self._map.get(key) + if lattice is None: + lattice = self._route(key) + return self._engine.apply(lattice, context, dimensions=dimensions) def apply_batch(self, contexts, **kwargs): """Partition contexts by CONTEXT_KEY fields; evaluate_batch per lattice.""" diff --git a/tests/accumulator/test_lattice.py b/tests/accumulator/test_lattice.py index 3921f86..9f7fe03 100644 --- a/tests/accumulator/test_lattice.py +++ b/tests/accumulator/test_lattice.py @@ -173,3 +173,144 @@ def test_load_missing_manifest_raises(self, tmp_path): (tmp_path / "empty").mkdir() with pytest.raises(FileNotFoundError): Lattice.load(tmp_path / "empty") + + +from pydantic import BaseModel + +from mountainash_rules import AmbiguousPartitionError +from mountainash_rules.engines.accumulator.engine import AccumulatorEngine +from mountainash_rules.engines.accumulator.aggregate import Aggregate +from mountainash_rules.core.constants import ( + UNKNOWN, + UNKNOWN_NUMERIC, + NOT_SET, + DimensionRole, + MatchStrategy, +) +from mountainash_rules.core.dimension import Dimension, DimensionsMetadata + + +class RoutingContext(BaseModel): + region: str | None = None + channel: str | None = None + product: str + + +def _routing_engine(): + metadata = DimensionsMetadata(dimensions=[ + Dimension( + dimension_name="region", + match_strategy=MatchStrategy.EXACT, + role=DimensionRole.CONTEXT_KEY, + ), + Dimension( + dimension_name="channel", + match_strategy=MatchStrategy.EXACT, + role=DimensionRole.CONTEXT_KEY, + ), + Dimension(dimension_name="product", match_strategy=MatchStrategy.EXACT), + ]) + return AccumulatorEngine( + dimension_metadata=metadata, + aggregates=[Aggregate(column_name="margin")], + ) + + +def _routing_rules(keys: list[tuple[str, str]]) -> pl.DataFrame: + """One rule per (region, channel) partition key; UNKNOWN = wildcard.""" + return pl.DataFrame({ + "region": [k[0] for k in keys], + "channel": [k[1] for k in keys], + "rule_name": [f"r{i}" for i in range(len(keys))], + "product": ["GOLD"] * len(keys), + "margin": [1.0] * len(keys), + }) + + +def _index_for(keys, validate=True): + engine = _routing_engine() + lattices = engine.build_all(_routing_rules(keys)) + return engine, engine.index(lattices, validate=validate) + + +class TestTernaryRouting: + def test_default_partition_catches_unmatched_context(self): + engine, index = _index_for([("AU", "BROKER"), (UNKNOWN, UNKNOWN)]) + result = index.apply(RoutingContext(region="NZ", channel="DIRECT", product="GOLD")) + assert result.count >= 1 # routed to the all-wildcard default + + def test_partial_specific_miss_falls_to_default(self): + engine, index = _index_for([("AU", "BROKER"), (UNKNOWN, UNKNOWN)]) + # Exercise _route (not the dict fast path): (AU, DIRECT) is not an + # exact key; (AU, BROKER) dies on channel; default survives. + result = index.apply(RoutingContext(region="AU", channel="DIRECT", product="GOLD")) + assert result.count >= 1 + + def test_exact_hit_uses_fast_path_and_wins(self): + engine, index = _index_for([("AU", "BROKER"), (UNKNOWN, UNKNOWN)]) + result = index.apply(RoutingContext(region="AU", channel="BROKER", product="GOLD")) + assert result.count >= 1 + + def test_runtime_ambiguity_raises(self): + engine, index = _index_for( + [("AU", UNKNOWN), (UNKNOWN, "BROKER")], validate=False + ) + with pytest.raises(AmbiguousPartitionError): + index.apply(RoutingContext(region="AU", channel="BROKER", product="GOLD")) + + def test_missing_key_field_routes_to_default(self): + engine, index = _index_for([("AU", "BROKER"), (UNKNOWN, UNKNOWN)]) + result = index.apply(RoutingContext(product="GOLD")) # region/channel None + assert result.count >= 1 + + def test_missing_key_field_without_default_raises(self): + engine, index = _index_for([("AU", "BROKER")]) + with pytest.raises(KeyError, match="No lattice"): + index.apply(RoutingContext(product="GOLD")) + + def test_context_unknown_sentinel_is_wildcard_only(self): + engine, index = _index_for([("AU", "BROKER"), (UNKNOWN, UNKNOWN)]) + result = index.apply( + RoutingContext(region=UNKNOWN, channel="BROKER", product="GOLD") + ) + # UNKNOWN region kills (AU, BROKER); only the default survives. + assert result.count >= 1 + + def test_wildcard_free_index_miss_raises_keyerror(self): + engine, index = _index_for([("AU", "BROKER")]) + with pytest.raises(KeyError, match="No lattice"): + index.apply(RoutingContext(region="US", channel="X", product="GOLD")) + + def test_ambiguous_is_a_keyerror(self): + assert issubclass(AmbiguousPartitionError, KeyError) + + +class TestIndexStructuralChecks: + def test_empty_index_raises(self): + engine = _routing_engine() + with pytest.raises(ValueError, match="at least one lattice"): + engine.index([]) + + def test_duplicate_keys_raise(self): + engine = _routing_engine() + lattices = engine.build_all(_routing_rules([("AU", "BROKER")])) + with pytest.raises(ValueError, match="[Dd]uplicate"): + engine.index(lattices + lattices) + + def test_not_set_key_raises(self): + engine = _routing_engine() + lattices = engine.build_all(_routing_rules([(NOT_SET, "BROKER")])) + with pytest.raises(ValueError, match="NOT_SET"): + engine.index(lattices) + + def test_keyless_lattice_with_key_dims_raises(self): + engine = _routing_engine() + (good,) = engine.build_all(_routing_rules([("AU", "BROKER")])) + flat = Lattice( + dataframe=good.combinations, + metadata=good.metadata, + aggregates=good.aggregates, + partition_key=None, + ) + with pytest.raises(ValueError, match="partition_key"): + engine.index([good, flat]) From 2d1877d360b12212d254ebb75faffafb32c4a010 Mon Sep 17 00:00:00 2001 From: Nathaniel Ramm Date: Sun, 19 Jul 2026 19:10:01 +1000 Subject: [PATCH 05/11] feat: route apply_batch combos through the shared ternary resolver --- .../engines/accumulator/lattice.py | 17 +++++---- tests/accumulator/test_lattice.py | 36 +++++++++++++++++++ 2 files changed, 47 insertions(+), 6 deletions(-) diff --git a/src/mountainash_rules/engines/accumulator/lattice.py b/src/mountainash_rules/engines/accumulator/lattice.py index 29f9ad4..6426fdc 100644 --- a/src/mountainash_rules/engines/accumulator/lattice.py +++ b/src/mountainash_rules/engines/accumulator/lattice.py @@ -254,13 +254,18 @@ def apply_batch(self, contexts, **kwargs): frames = [] n = len(combos[key_fields[0]]) for i in range(n): - key = tuple(combos[f][i] for f in key_fields) - if key not in self._map: - raise KeyError(f"No lattice for partition key {key!r}") + raw_key = tuple(combos[f][i] for f in key_fields) + key = self._engine._normalize_partition_key(raw_key) + lattice = self._map.get(key) + if lattice is None: + lattice = self._route(key) part = rel - for f, v in zip(key_fields, key): - part = part.filter(ma.col(f).eq(ma.lit(v))) - engine = self._engine._filter_engine_for(self._map[key]) + for f, v in zip(key_fields, raw_key): + part = part.filter( + ma.col(f).is_null() if v is None + else ma.col(f).eq(ma.lit(v)) + ) + engine = self._engine._filter_engine_for(lattice) frames.append(relation(engine.evaluate_batch( part.collect(), **kwargs ).survivors)) diff --git a/tests/accumulator/test_lattice.py b/tests/accumulator/test_lattice.py index 9f7fe03..05cbbfe 100644 --- a/tests/accumulator/test_lattice.py +++ b/tests/accumulator/test_lattice.py @@ -314,3 +314,39 @@ def test_keyless_lattice_with_key_dims_raises(self): ) with pytest.raises(ValueError, match="partition_key"): engine.index([good, flat]) + + +class TestTernaryRoutingBatch: + def test_batch_mixes_specific_and_default(self): + engine, index = _index_for([("AU", "BROKER"), (UNKNOWN, UNKNOWN)]) + contexts = pl.DataFrame({ + "region": ["AU", "NZ", None], + "channel": ["BROKER", "DIRECT", None], + "product": ["GOLD", "GOLD", "GOLD"], + }) + result = index.apply_batch(contexts) + rows = relation(result.survivors).to_dict() + # every context produced at least one survivor row + assert set(rows["__context_id"]) == {0, 1, 2} + + def test_batch_unroutable_combo_raises(self): + engine, index = _index_for([("AU", "BROKER")]) + contexts = pl.DataFrame({ + "region": ["AU", "US"], + "channel": ["BROKER", "X"], + "product": ["GOLD", "GOLD"], + }) + with pytest.raises(KeyError, match="No lattice"): + index.apply_batch(contexts) + + def test_batch_ambiguous_combo_raises(self): + engine, index = _index_for( + [("AU", UNKNOWN), (UNKNOWN, "BROKER")], validate=False + ) + contexts = pl.DataFrame({ + "region": ["AU"], + "channel": ["BROKER"], + "product": ["GOLD"], + }) + with pytest.raises(AmbiguousPartitionError): + index.apply_batch(contexts) From 5a81293c5023e184ce5b1066b512f256bff4eed1 Mon Sep 17 00:00:00 2001 From: Nathaniel Ramm Date: Sun, 19 Jul 2026 19:17:58 +1000 Subject: [PATCH 06/11] feat: exhaustive load-time ambiguity validation for lattice indexes --- .../engines/accumulator/engine.py | 16 +++- .../engines/accumulator/lattice.py | 95 ++++++++++++++++++- tests/accumulator/test_lattice.py | 65 +++++++++++++ 3 files changed, 173 insertions(+), 3 deletions(-) diff --git a/src/mountainash_rules/engines/accumulator/engine.py b/src/mountainash_rules/engines/accumulator/engine.py index d392177..02923fd 100644 --- a/src/mountainash_rules/engines/accumulator/engine.py +++ b/src/mountainash_rules/engines/accumulator/engine.py @@ -548,9 +548,21 @@ def index( validate: bool = True, max_witnesses: int = 1_000_000, ) -> LatticeIndex: - """Build a partition-key routing index over pre-built lattices.""" + """Build a partition-key routing index over pre-built lattices. + + Args: + lattices: List of Lattice objects from build_all() or load(). + validate: Run the exhaustive load-time ambiguity check + (structural checks — empty/duplicate/NOT_SET keys — run + regardless). + max_witnesses: Ceiling on the validation matrix size; above + it index() raises ValueError rather than sampling. + """ from mountainash_rules.engines.accumulator.lattice import LatticeIndex - return LatticeIndex(self, lattices, self._context_key_dims) + return LatticeIndex( + self, lattices, self._context_key_dims, + validate=validate, max_witnesses=max_witnesses, + ) def _extract_partition_key(self, context: t.Any) -> tuple: """Extract the partition key tuple from a context object. diff --git a/src/mountainash_rules/engines/accumulator/lattice.py b/src/mountainash_rules/engines/accumulator/lattice.py index 6426fdc..f905fd9 100644 --- a/src/mountainash_rules/engines/accumulator/lattice.py +++ b/src/mountainash_rules/engines/accumulator/lattice.py @@ -1,3 +1,4 @@ +import itertools import pathlib import typing as t import yaml @@ -10,6 +11,7 @@ HitPolicy, MatchStrategy, not_set_sentinel_for, + unknown_sentinel_for, ) from mountainash_rules.core.dimension import Dimension, DimensionsMetadata from mountainash_rules.engines.filter.engine import ExpressionRulesEngine @@ -124,7 +126,14 @@ class LatticeIndex: O(1) dict fast path. """ - def __init__(self, engine, lattices: list["Lattice"], context_key_dims) -> None: + def __init__( + self, + engine, + lattices: list["Lattice"], + context_key_dims, + validate: bool = True, + max_witnesses: int = 1_000_000, + ) -> None: self._engine = engine self._context_key_dims = list(context_key_dims) self._lattices = list(lattices) @@ -164,6 +173,9 @@ def __init__(self, engine, lattices: list["Lattice"], context_key_dims) -> None: self._build_meta_engine() if self._context_key_dims else None ) + if validate and self._meta_engine is not None: + self._validate_ambiguity(max_witnesses) + def _build_meta_engine(self) -> "ExpressionRulesEngine": """One meta-rule row per partition; EXACT_KEY per key dim.""" columns: dict[str, list] = { @@ -193,6 +205,87 @@ def _build_meta_engine(self) -> "ExpressionRulesEngine": dimension_metadata=meta_metadata, ) + _WITNESS_CHUNK = 100_000 + + def _validate_ambiguity(self, max_witnesses: int) -> None: + """Exhaustive witness-matrix ambiguity check (spec §5). + + Per key dim the reachable context values collapse into finitely + many equivalence classes: each specific key value, plus OTHER — + represented by the typed NOT_SET sentinel (None for bool), which + matches no specific key and every wildcard. The cross-product of + classes is routed through the meta-engine in chunks; any witness + with >= 2 top-specificity survivors is a reachable runtime tie. + """ + classes: list[list] = [] + for i, d in enumerate(self._context_key_dims): + if d.data_type is DataType.BOOL: + wildcard, other = None, None + else: + wildcard = unknown_sentinel_for(d.data_type) + other = not_set_sentinel_for(d.data_type) + specifics = sorted( + {key[i] for key in self._map if key[i] != wildcard}, + key=repr, + ) + classes.append(specifics + [other]) + + total = 1 + for c in classes: + total *= len(c) + if total > max_witnesses: + raise ValueError( + f"Ambiguity validation needs {total} witness contexts, " + f"over max_witnesses={max_witnesses}; pass a higher " + f"max_witnesses, validate=False (accepting the runtime " + f"tie check), or restructure the key dimensions" + ) + + fields = [d.resolved_context_field for d in self._context_key_dims] + witnesses = itertools.product(*classes) + while True: + chunk = list(itertools.islice(witnesses, self._WITNESS_CHUNK)) + if not chunk: + return + contexts = relation({ + "__witness_id": list(range(len(chunk))), + **{ + f: [w[i] for w in chunk] + for i, f in enumerate(fields) + }, + }).collect() + survivors = relation( + self._meta_engine.evaluate_batch( + contexts, context_id_field="__witness_id" + ).survivors + ).to_dict() + best: dict[int, int] = {} + tied: dict[int, list[int]] = {} + for wid, spec, pidx in zip( + survivors["__context_id"], + survivors["__specificity"], + survivors["__partition_idx"], + ): + if wid not in best or spec > best[wid]: + best[wid] = spec + tied[wid] = [pidx] + elif spec == best[wid]: + tied[wid].append(pidx) + for wid, parts in tied.items(): + if len(parts) > 1: + witness_ctx = dict(zip(fields, chunk[wid])) + tied_keys = [ + tuple( + self._lattices[p].partition_key[d.dimension_name] + for d in self._context_key_dims + ) + for p in parts + ] + raise AmbiguousPartitionError( + f"Partition suite is ambiguous: witness context " + f"{witness_ctx!r} ties partitions {tied_keys!r}" + ) + def _route(self, key: tuple) -> "Lattice": """Ternary + specificity routing for a normalised key tuple.""" ctx = { diff --git a/tests/accumulator/test_lattice.py b/tests/accumulator/test_lattice.py index 05cbbfe..28c691b 100644 --- a/tests/accumulator/test_lattice.py +++ b/tests/accumulator/test_lattice.py @@ -350,3 +350,68 @@ def test_batch_ambiguous_combo_raises(self): }) with pytest.raises(AmbiguousPartitionError): index.apply_batch(contexts) + + +class TestIndexValidation: + def test_crossing_pair_without_cover_raises_at_index(self): + engine = _routing_engine() + lattices = engine.build_all( + _routing_rules([("AU", UNKNOWN), (UNKNOWN, "BROKER")]) + ) + with pytest.raises(AmbiguousPartitionError) as exc: + engine.index(lattices) + # message carries a witness context + assert "AU" in str(exc.value) and "BROKER" in str(exc.value) + + def test_crossing_pair_with_cover_validates_and_routes(self): + engine = _routing_engine() + lattices = engine.build_all(_routing_rules([ + ("AU", UNKNOWN), (UNKNOWN, "BROKER"), ("AU", "BROKER"), + ])) + index = engine.index(lattices) # must NOT raise (false-positive guard) + result = index.apply( + RoutingContext(region="AU", channel="BROKER", product="GOLD") + ) + assert result.count >= 1 + + def test_validate_false_defers_to_runtime(self): + engine = _routing_engine() + lattices = engine.build_all( + _routing_rules([("AU", UNKNOWN), (UNKNOWN, "BROKER")]) + ) + index = engine.index(lattices, validate=False) # no raise here + with pytest.raises(AmbiguousPartitionError): + index.apply(RoutingContext(region="AU", channel="BROKER", product="GOLD")) + + def test_witness_cap_overflow_raises(self): + engine = _routing_engine() + lattices = engine.build_all( + _routing_rules([("AU", "BROKER"), ("NZ", "DIRECT")]) + ) + # 3 classes per dim (AU, NZ, OTHER) x (BROKER, DIRECT, OTHER) = 9 > 4 + with pytest.raises(ValueError, match="max_witnesses"): + engine.index(lattices, max_witnesses=4) + + def test_bool_key_dim_full_domain_validates(self): + metadata = DimensionsMetadata(dimensions=[ + Dimension( + dimension_name="flag", + match_strategy=MatchStrategy.EXACT, + data_type="bool", + role=DimensionRole.CONTEXT_KEY, + ), + Dimension(dimension_name="product", match_strategy=MatchStrategy.EXACT), + ]) + engine = AccumulatorEngine( + dimension_metadata=metadata, + aggregates=[Aggregate(column_name="margin")], + ) + rules = pl.DataFrame({ + "flag": [True, False, None], # null rule value = bool wildcard + "rule_name": ["r0", "r1", "r2"], + "product": ["GOLD"] * 3, + "margin": [1.0] * 3, + }) + index = engine.index(engine.build_all(rules)) # OTHER = None, no raise + result = index.apply({"product": "GOLD"}) # flag missing -> wildcard + assert result.count >= 1 From 606991f3792d37c3927efb7c4881cf8066e73c91 Mon Sep 17 00:00:00 2001 From: Nathaniel Ramm Date: Sun, 19 Jul 2026 19:27:34 +1000 Subject: [PATCH 07/11] fix: narrow Optional partition_key/meta_engine accesses (mypy) --- .../engines/accumulator/lattice.py | 38 +++++++++++-------- 1 file changed, 23 insertions(+), 15 deletions(-) diff --git a/src/mountainash_rules/engines/accumulator/lattice.py b/src/mountainash_rules/engines/accumulator/lattice.py index f905fd9..de96e64 100644 --- a/src/mountainash_rules/engines/accumulator/lattice.py +++ b/src/mountainash_rules/engines/accumulator/lattice.py @@ -183,10 +183,10 @@ def _build_meta_engine(self) -> "ExpressionRulesEngine": } columns["__partition_idx"] = [] for idx, lattice in enumerate(self._lattices): + pk = lattice.partition_key + assert pk is not None # key dims present ⇒ structural check guarantees this for d in self._context_key_dims: - columns[d.dimension_name].append( - lattice.partition_key[d.dimension_name] - ) + columns[d.dimension_name].append(pk[d.dimension_name]) columns["__partition_idx"].append(idx) meta_metadata = DimensionsMetadata( dimensions=[ @@ -217,6 +217,7 @@ def _validate_ambiguity(self, max_witnesses: int) -> None: classes is routed through the meta-engine in chunks; any witness with >= 2 top-specificity survivors is a reachable runtime tie. """ + assert self._meta_engine is not None # only called when key dims exist classes: list[list] = [] for i, d in enumerate(self._context_key_dims): if d.data_type is DataType.BOOL: @@ -274,13 +275,16 @@ def _validate_ambiguity(self, max_witnesses: int) -> None: for wid, parts in tied.items(): if len(parts) > 1: witness_ctx = dict(zip(fields, chunk[wid])) - tied_keys = [ - tuple( - self._lattices[p].partition_key[d.dimension_name] - for d in self._context_key_dims + tied_keys = [] + for p in parts: + pk = self._lattices[p].partition_key + assert pk is not None + tied_keys.append( + tuple( + pk[d.dimension_name] + for d in self._context_key_dims + ) ) - for p in parts - ] raise AmbiguousPartitionError( f"Partition suite is ambiguous: witness context " f"{witness_ctx!r} ties partitions {tied_keys!r}" @@ -292,6 +296,7 @@ def _route(self, key: tuple) -> "Lattice": d.resolved_context_field: key[i] for i, d in enumerate(self._context_key_dims) } + assert self._meta_engine is not None # dict miss with key dims ⇒ non-None rows = relation(self._meta_engine.evaluate(ctx).survivors).to_dict() idxs = rows["__partition_idx"] if not idxs: @@ -304,13 +309,16 @@ def _route(self, key: tuple) -> "Lattice": i for i, s in zip(idxs, rows["__specificity"]) if s == top ] if len(tied) > 1: - tied_keys = [ - tuple( - self._lattices[i].partition_key[d.dimension_name] - for d in self._context_key_dims + tied_keys = [] + for i in tied: + pk = self._lattices[i].partition_key + assert pk is not None + tied_keys.append( + tuple( + pk[d.dimension_name] + for d in self._context_key_dims + ) ) - for i in tied - ] raise AmbiguousPartitionError( f"Context key {key!r} ties {len(tied)} partitions at " f"specificity {top}: {tied_keys!r}" From d50cbbfb868262a95626b4b79c85c962789d1cc7 Mon Sep 17 00:00:00 2001 From: Nathaniel Ramm Date: Sun, 19 Jul 2026 19:32:16 +1000 Subject: [PATCH 08/11] test: routing persistence round-trip; docs: ternary partition routing Co-Authored-By: Claude Opus 4.8 (1M context) --- CLAUDE.md | 3 ++- tests/accumulator/test_lattice.py | 17 +++++++++++++++++ 2 files changed, 19 insertions(+), 1 deletion(-) diff --git a/CLAUDE.md b/CLAUDE.md index 6d44b19..55becd1 100644 --- a/CLAUDE.md +++ b/CLAUDE.md @@ -46,7 +46,7 @@ no hit-policy interaction. Shares Steps 1–3 with `evaluate` via - `build(rules)` → `Lattice`: level-wise expansion of compatible rule combinations, coalescing dimension values (`co_` columns + `co_*_na` flags, `accumulator_compiler.py`), accumulating `Aggregate` numerics (`__agg_`), identifying combinations by prime products (`primes.py`; overflow → `LatticeWidthExceededError` with remediation), and pruning dominated combinations (frontier filter). Each level is **materialised** via `relation(...).collect()` — do not re-chain lazily (exponential plan blow-up). - `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. +- `DimensionRole.CONTEXT_KEY` dimensions partition the rule space; `build_all` + `index(lattices)` → `LatticeIndex` routes contexts (single or batch) to the right lattice using ternary + specificity semantics via an embedded meta-engine (`EXACT_KEY` dims): an all-wildcard key is the default/overflow partition, exact hits keep an O(1) dict fast path, ties raise `AmbiguousPartitionError` (a `KeyError` subclass, exported from the package root). `index(lattices, validate=True, max_witnesses=1_000_000)` runs an exhaustive witness-matrix ambiguity check at load time; empty/duplicate/NOT_SET-bearing keys are always rejected. See `docs/superpowers/specs/2026-07-19-ternary-partition-routing-design.md`. - `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`. @@ -61,6 +61,7 @@ No module under `src/mountainash_rules/` may import polars/ibis/narwhals directl | Strategy | Rule column format | Notes | |---|---|---| | `exact` / `not_equal` | scalar | any data_type | +| `exact_key` | scalar | rule-side wildcard only (UNKNOWN → 0; context sentinel vs specific → −1); powers partition routing | | `range` | two columns (`range_min_field`/`range_max_field`) | numeric/temporal; `range_min_inclusive`/`range_max_inclusive` flags | | `greater_than` / `less_than` | threshold | numeric/temporal | | `prefix` / `suffix` / `contains` | string | | diff --git a/tests/accumulator/test_lattice.py b/tests/accumulator/test_lattice.py index 28c691b..85a09c9 100644 --- a/tests/accumulator/test_lattice.py +++ b/tests/accumulator/test_lattice.py @@ -415,3 +415,20 @@ def test_bool_key_dim_full_domain_validates(self): index = engine.index(engine.build_all(rules)) # OTHER = None, no raise result = index.apply({"product": "GOLD"}) # flag missing -> wildcard assert result.count >= 1 + + +class TestRoutingPersistence: + def test_saved_wildcard_suite_routes_after_load(self, tmp_path): + engine = _routing_engine() + built = engine.build_all( + _routing_rules([("AU", "BROKER"), (UNKNOWN, UNKNOWN)]) + ) + loaded = [ + Lattice.load(lattice.save(tmp_path / f"part{i}")) + for i, lattice in enumerate(built) + ] + index = engine.index(loaded) + result = index.apply( + RoutingContext(region="NZ", channel="DIRECT", product="GOLD") + ) + assert result.count >= 1 # sentinel key survived the round-trip From 581270a3ad60d0675a9b931c4b499d8b46707cfa Mon Sep 17 00:00:00 2001 From: Nathaniel Ramm Date: Sun, 19 Jul 2026 19:49:16 +1000 Subject: [PATCH 09/11] fix: KeyError contract on bool-keyed partition miss; persistence + doc coverage - sorted(self._map, key=repr) so a bool wildcard (None) in served keys can't raise TypeError and mask the KeyError (breaks the KeyError->422 contract) - persistence round-trip tests for numeric + bool wildcard sentinels - document bool EXACT_KEY string-coalesce reliance --- src/mountainash_rules/core/compiler.py | 6 ++ .../engines/accumulator/lattice.py | 2 +- tests/accumulator/test_lattice.py | 98 +++++++++++++++++++ 3 files changed, 105 insertions(+), 1 deletion(-) diff --git a/src/mountainash_rules/core/compiler.py b/src/mountainash_rules/core/compiler.py index a7d5e6a..e752948 100644 --- a/src/mountainash_rules/core/compiler.py +++ b/src/mountainash_rules/core/compiler.py @@ -81,6 +81,12 @@ def _compile_exact_key(self, dim: Dimension) -> BaseExpressionAPI: rule_col = ma.col(dim.resolved_rule_field) ctx_col = ma.col(CTX_PREFIX + dim.dimension_name) if dim.data_type is DataType.BOOL: + # Bool: rule null is the wildcard (checked above), unlike other + # data types the rule-side wildcard has no sentinel value to + # compare against. The equality check below then compares + # rule_col and ctx_col as native Booleans directly (both are + # true/false/null here, never the string sentinels) — there is + # no stringification involved for this data type. wildcard = rule_col.is_null() else: wildcard = rule_col.__eq__( diff --git a/src/mountainash_rules/engines/accumulator/lattice.py b/src/mountainash_rules/engines/accumulator/lattice.py index de96e64..3d8e16b 100644 --- a/src/mountainash_rules/engines/accumulator/lattice.py +++ b/src/mountainash_rules/engines/accumulator/lattice.py @@ -302,7 +302,7 @@ def _route(self, key: tuple) -> "Lattice": if not idxs: raise KeyError( f"No lattice for partition key {key!r}; served partition " - f"keys: {sorted(self._map)!r}" + f"keys: {sorted(self._map, key=repr)!r}" ) top = rows["__specificity"][0] # survivors are rank-sorted tied = [ diff --git a/tests/accumulator/test_lattice.py b/tests/accumulator/test_lattice.py index 85a09c9..c070762 100644 --- a/tests/accumulator/test_lattice.py +++ b/tests/accumulator/test_lattice.py @@ -432,3 +432,101 @@ def test_saved_wildcard_suite_routes_after_load(self, tmp_path): RoutingContext(region="NZ", channel="DIRECT", product="GOLD") ) assert result.count >= 1 # sentinel key survived the round-trip + + def test_saved_numeric_wildcard_suite_routes_after_load(self, tmp_path): + metadata = DimensionsMetadata(dimensions=[ + Dimension( + dimension_name="tier", + match_strategy=MatchStrategy.EXACT, + data_type="int", + role=DimensionRole.CONTEXT_KEY, + ), + Dimension(dimension_name="product", match_strategy=MatchStrategy.EXACT), + ]) + engine = AccumulatorEngine( + dimension_metadata=metadata, + aggregates=[Aggregate(column_name="margin")], + ) + rules = pl.DataFrame({ + "tier": [1, UNKNOWN_NUMERIC], # specific + numeric wildcard + "rule_name": ["r0", "r1"], + "product": ["GOLD", "GOLD"], + "margin": [1.0, 1.0], + }) + built = engine.build_all(rules) + loaded = [ + Lattice.load(lattice.save(tmp_path / f"part{i}")) + for i, lattice in enumerate(built) + ] + index = engine.index(loaded) + result = index.apply({"tier": 999, "product": "GOLD"}) # -> wildcard + assert result.count >= 1 + + def test_saved_bool_wildcard_suite_routes_after_load(self, tmp_path): + metadata = DimensionsMetadata(dimensions=[ + Dimension( + dimension_name="flag", + match_strategy=MatchStrategy.EXACT, + data_type="bool", + role=DimensionRole.CONTEXT_KEY, + ), + Dimension(dimension_name="product", match_strategy=MatchStrategy.EXACT), + ]) + engine = AccumulatorEngine( + dimension_metadata=metadata, + aggregates=[Aggregate(column_name="margin")], + ) + rules = pl.DataFrame({ + "flag": [True, None], # specific + bool wildcard (null) + "rule_name": ["r0", "r1"], + "product": ["GOLD", "GOLD"], + "margin": [1.0, 1.0], + }) + built = engine.build_all(rules) + loaded = [ + Lattice.load(lattice.save(tmp_path / f"part{i}")) + for i, lattice in enumerate(built) + ] + index = engine.index(loaded) + result = index.apply({"flag": False, "product": "GOLD"}) # -> wildcard + assert result.count >= 1 + + +class TestRoutingErrorContract: + """A no-survivor miss must raise KeyError (not TypeError) even when a + bool key dim puts None alongside non-None values in the served keys — + so the service's `except KeyError -> 422` handler still catches it.""" + + def test_bool_wildcard_miss_raises_keyerror_not_typeerror(self): + metadata = DimensionsMetadata(dimensions=[ + Dimension( + dimension_name="region", + match_strategy=MatchStrategy.EXACT, + role=DimensionRole.CONTEXT_KEY, + ), + Dimension( + dimension_name="flag", + match_strategy=MatchStrategy.EXACT, + data_type="bool", + role=DimensionRole.CONTEXT_KEY, + ), + Dimension(dimension_name="product", match_strategy=MatchStrategy.EXACT), + ]) + engine = AccumulatorEngine( + dimension_metadata=metadata, + aggregates=[Aggregate(column_name="margin")], + ) + # Two partitions sharing region 'AU': one specific flag, one bool + # wildcard (null). served-keys list therefore mixes None and True. + rules = pl.DataFrame({ + "region": ["AU", "AU"], + "flag": [True, None], + "rule_name": ["r0", "r1"], + "product": ["GOLD", "GOLD"], + "margin": [1.0, 1.0], + }) + index = engine.index(engine.build_all(rules)) # validate=True default + # region 'US' misses both partitions -> no survivor -> the served-keys + # message sorts {('AU', True), ('AU', None)}; must not TypeError. + with pytest.raises(KeyError, match="No lattice"): + index.apply({"region": "US", "flag": True, "product": "GOLD"}) From 109e20f8e0da7e5ddc4e60c1175e697d53cb7586 Mon Sep 17 00:00:00 2001 From: Nathaniel Ramm Date: Sun, 19 Jul 2026 20:41:52 +1000 Subject: [PATCH 10/11] fix: batch/NaN/apply_auto parity from Codex adversarial review - _normalize_partition_key treats float NaN as missing (spec: null/NaN == absent) - apply_batch normalises key columns up front: absent columns filled, null/NaN coalesced to typed NOT_SET (bool keeps null); routing now mirrors single apply and no missing/NaN row is silently dropped - apply_auto uses validate=False so the witness matrix isn't rerun per call --- .../engines/accumulator/engine.py | 28 +++++-- .../engines/accumulator/lattice.py | 75 ++++++++++++++++--- tests/accumulator/test_apply.py | 54 +++++++++++++ tests/accumulator/test_lattice.py | 40 ++++++++++ 4 files changed, 179 insertions(+), 18 deletions(-) diff --git a/src/mountainash_rules/engines/accumulator/engine.py b/src/mountainash_rules/engines/accumulator/engine.py index 02923fd..fe039d3 100644 --- a/src/mountainash_rules/engines/accumulator/engine.py +++ b/src/mountainash_rules/engines/accumulator/engine.py @@ -585,12 +585,21 @@ def _extract_partition_key(self, context: t.Any) -> tuple: )) def _normalize_partition_key(self, key: tuple) -> tuple: - """Map None key values to the typed NOT_SET sentinel (None for bool).""" - return tuple( - v if v is not None or d.data_type is DataType.BOOL - else not_set_sentinel_for(d.data_type) - for v, d in zip(key, self._context_key_dims) - ) + """Map missing key values to the typed NOT_SET sentinel (None for bool). + + A value counts as missing when it is None or a float NaN — backend + nulls and NaN are treated identically to an absent field (spec §1). + """ + out = [] + for v, d in zip(key, self._context_key_dims): + missing = v is None or (isinstance(v, float) and v != v) + if not missing: + out.append(v) + elif d.data_type is DataType.BOOL: + out.append(None) + else: + out.append(not_set_sentinel_for(d.data_type)) + return tuple(out) def apply_auto( self, @@ -613,5 +622,10 @@ def apply_auto( Note: Convenience wrapper; hot paths should hold a LatticeIndex. + Load-time ambiguity validation is skipped here (it would rerun + the witness matrix every call); routing still raises + AmbiguousPartitionError at apply time on a genuine tie. """ - return self.index(lattices).apply(context, dimensions=dimensions) + return self.index(lattices, validate=False).apply( + context, dimensions=dimensions + ) diff --git a/src/mountainash_rules/engines/accumulator/lattice.py b/src/mountainash_rules/engines/accumulator/lattice.py index 3d8e16b..72d24ba 100644 --- a/src/mountainash_rules/engines/accumulator/lattice.py +++ b/src/mountainash_rules/engines/accumulator/lattice.py @@ -348,28 +348,81 @@ def apply_batch(self, contexts, **kwargs): # Synthesise a GLOBAL context id before partitioning: per-partition # batches would otherwise restart ids at 0 and collide after concat. if not kwargs.get("context_id_field"): - rel = rel.with_row_index(name="__lattice_ctx_id") + # relation(...collect()) (not just .collect()): `.columns` on a + # lazy relation doesn't reflect a just-added with_row_index + # column until materialised, and `passthrough` below reads + # `.columns`. A bare `.collect()` also degrades to a raw + # DataFrame that no longer dispatches with_columns() through the + # mountainash expression layer — re-wrapping keeps that intact. + rel = relation(rel.with_row_index(name="__lattice_ctx_id").collect()) kwargs["context_id_field"] = "__lattice_ctx_id" - combos = rel.select(*[ma.col(f) for f in key_fields]).unique().to_dict() + # Normalise the key columns once so batch routing mirrors single + # apply(): a key column absent from the whole batch is treated as + # missing; null/NaN coalesce to the typed NOT_SET sentinel (bool keeps + # null — its don't-care). Combo grouping AND per-partition filtering + # both use these normalised columns, so a missing/NaN context routes + # exactly as _extract_partition_key would, and no row is silently + # dropped or double-counted. + passthrough = list(rel.columns) + existing = set(passthrough) + norm_fields: list[str] = [] + norm_exprs = [] + for d in self._context_key_dims: + f = d.resolved_context_field + nf = "__key_" + f + norm_fields.append(nf) + if f not in existing: + fill = ( + None + if d.data_type is DataType.BOOL + else not_set_sentinel_for(d.data_type) + ) + norm_exprs.append(ma.lit(fill).alias(nf)) + elif d.data_type is DataType.BOOL: + norm_exprs.append(ma.col(f).alias(nf)) + elif d.data_type is DataType.FLOAT: + sentinel = not_set_sentinel_for(d.data_type) + norm_exprs.append( + ma.when(ma.col(f).is_null() | ma.col(f).is_nan()) + .then(ma.lit(sentinel)) + .otherwise(ma.col(f)) + .alias(nf) + ) + else: + sentinel = not_set_sentinel_for(d.data_type) + norm_exprs.append( + ma.coalesce(ma.col(f), ma.lit(sentinel)).alias(nf) + ) + rel = rel.with_columns(*norm_exprs) + + combos = ( + rel.select(*[ma.col(nf) for nf in norm_fields]).unique().to_dict() + ) frames = [] - n = len(combos[key_fields[0]]) + n = len(combos[norm_fields[0]]) for i in range(n): - raw_key = tuple(combos[f][i] for f in key_fields) - key = self._engine._normalize_partition_key(raw_key) + # Combos are already normalised; only bool positions can be None. + key = tuple(combos[nf][i] for nf in norm_fields) lattice = self._map.get(key) if lattice is None: lattice = self._route(key) part = rel - for f, v in zip(key_fields, raw_key): + for nf, v in zip(norm_fields, key): part = part.filter( - ma.col(f).is_null() if v is None - else ma.col(f).eq(ma.lit(v)) + ma.col(nf).is_null() + if v is None + else ma.col(nf).eq(ma.lit(v)) ) + # Drop the transient __key_* columns so the filter engine sees the + # original context frame unchanged. + part = part.select(*[ma.col(c) for c in passthrough]) engine = self._engine._filter_engine_for(lattice) - frames.append(relation(engine.evaluate_batch( - part.collect(), **kwargs - ).survivors)) + frames.append( + relation( + engine.evaluate_batch(part.collect(), **kwargs).survivors + ) + ) merged = concat(frames).collect() from mountainash_rules.core.batch_result import BatchRuleResult diff --git a/tests/accumulator/test_apply.py b/tests/accumulator/test_apply.py index da1dc1c..6ff3f37 100644 --- a/tests/accumulator/test_apply.py +++ b/tests/accumulator/test_apply.py @@ -192,3 +192,57 @@ def test_normalize_partition_key(self): engine = self._engine() assert engine._normalize_partition_key((None,)) == (NOT_SET_NUMERIC,) assert engine._normalize_partition_key((7,)) == (7,) + + +class TestApplyAutoAndNaN: + def _float_engine(self): + metadata = DimensionsMetadata(dimensions=[ + Dimension( + dimension_name="score", + match_strategy=MatchStrategy.EXACT, + data_type="float", + role=DimensionRole.CONTEXT_KEY, + ), + Dimension(dimension_name="channel", match_strategy=MatchStrategy.EXACT), + ]) + return AccumulatorEngine( + dimension_metadata=metadata, + aggregates=[Aggregate(column_name="margin")], + ) + + def test_nan_key_field_normalizes_to_not_set(self): + engine = self._float_engine() + key = engine._extract_partition_key({"score": float("nan"), "channel": "BROKER"}) + assert key == (NOT_SET_NUMERIC,) + + def test_normalize_partition_key_handles_nan(self): + engine = self._float_engine() + assert engine._normalize_partition_key((float("nan"),)) == (NOT_SET_NUMERIC,) + assert engine._normalize_partition_key((2.5,)) == (2.5,) + + def test_apply_auto_skips_load_validation(self): + # A crossing pair (AU,*)/(*,BROKER) is ambiguous -> index(validate=True) + # would raise. apply_auto must not pay that: a context that routes + # unambiguously still succeeds. + metadata = DimensionsMetadata(dimensions=[ + Dimension(dimension_name="region", match_strategy=MatchStrategy.EXACT, role=DimensionRole.CONTEXT_KEY), + Dimension(dimension_name="channel", match_strategy=MatchStrategy.EXACT, role=DimensionRole.CONTEXT_KEY), + Dimension(dimension_name="product", match_strategy=MatchStrategy.EXACT), + ]) + engine = AccumulatorEngine( + dimension_metadata=metadata, + aggregates=[Aggregate(column_name="margin")], + ) + rules = pl.DataFrame({ + "region": ["AU", UNKNOWN], + "channel": [UNKNOWN, "BROKER"], + "rule_name": ["r0", "r1"], + "product": ["GOLD", "GOLD"], + "margin": [1.0, 1.0], + }) + lattices = engine.build_all(rules) + # (AU, DIRECT): only (AU,*) survives -> unambiguous route, no raise. + result = engine.apply_auto( + lattices, {"region": "AU", "channel": "DIRECT", "product": "GOLD"} + ) + assert result.count >= 1 diff --git a/tests/accumulator/test_lattice.py b/tests/accumulator/test_lattice.py index c070762..391feb3 100644 --- a/tests/accumulator/test_lattice.py +++ b/tests/accumulator/test_lattice.py @@ -351,6 +351,46 @@ def test_batch_ambiguous_combo_raises(self): with pytest.raises(AmbiguousPartitionError): index.apply_batch(contexts) + def test_batch_absent_key_column_routes_to_default(self): + # region & channel columns entirely absent -> both rows treated as + # missing (NOT_SET) -> routed to the all-wildcard default, mirroring + # single apply(); must not raise a column-not-found error. + engine, index = _index_for([("AU", "BROKER"), (UNKNOWN, UNKNOWN)]) + contexts = pl.DataFrame({"product": ["GOLD", "GOLD"]}) + result = index.apply_batch(contexts) + rows = relation(result.survivors).to_dict() + assert set(rows["__context_id"]) == {0, 1} + + def test_batch_nan_key_value_routes_to_default_not_dropped(self): + metadata = DimensionsMetadata(dimensions=[ + Dimension( + dimension_name="score", + match_strategy=MatchStrategy.EXACT, + data_type="float", + role=DimensionRole.CONTEXT_KEY, + ), + Dimension(dimension_name="product", match_strategy=MatchStrategy.EXACT), + ]) + engine = AccumulatorEngine( + dimension_metadata=metadata, + aggregates=[Aggregate(column_name="margin")], + ) + rules = pl.DataFrame({ + "score": [1.5, float(UNKNOWN_NUMERIC)], # specific + float wildcard + "rule_name": ["r0", "r1"], + "product": ["GOLD", "GOLD"], + "margin": [1.0, 1.0], + }) + index = engine.index(engine.build_all(rules)) + contexts = pl.DataFrame({ + "score": [1.5, float("nan")], # second row NaN -> NOT_SET -> default + "product": ["GOLD", "GOLD"], + }) + result = index.apply_batch(contexts) + rows = relation(result.survivors).to_dict() + # BOTH context ids present: the NaN row must be routed, not silently dropped. + assert set(rows["__context_id"]) == {0, 1} + class TestIndexValidation: def test_crossing_pair_without_cover_raises_at_index(self): From 4a2a9c08aa19477d793ecb2b8d866927f19d502d Mon Sep 17 00:00:00 2001 From: Nathaniel Ramm Date: Sun, 19 Jul 2026 20:50:59 +1000 Subject: [PATCH 11/11] test: cover caller-supplied context_id_field in LatticeIndex.apply_batch Closes the final-review coverage gap on the provided-id branch (Codex-fix review). Co-Authored-By: Claude Opus 4.8 (1M context) --- tests/accumulator/test_lattice.py | 15 +++++++++++++++ 1 file changed, 15 insertions(+) diff --git a/tests/accumulator/test_lattice.py b/tests/accumulator/test_lattice.py index 391feb3..2cef8d2 100644 --- a/tests/accumulator/test_lattice.py +++ b/tests/accumulator/test_lattice.py @@ -391,6 +391,21 @@ def test_batch_nan_key_value_routes_to_default_not_dropped(self): # BOTH context ids present: the NaN row must be routed, not silently dropped. assert set(rows["__context_id"]) == {0, 1} + def test_batch_respects_caller_context_id_field(self): + # The caller-supplied context_id_field branch skips the internal + # __lattice_ctx_id synthesis; the caller's ids must still survive + # routing (specific + default + missing) through a partitioned index. + engine, index = _index_for([("AU", "BROKER"), (UNKNOWN, UNKNOWN)]) + contexts = pl.DataFrame({ + "my_id": [100, 200, 300], + "region": ["AU", "NZ", None], + "channel": ["BROKER", "DIRECT", None], + "product": ["GOLD", "GOLD", "GOLD"], + }) + result = index.apply_batch(contexts, context_id_field="my_id") + rows = relation(result.survivors).to_dict() + assert set(rows["__context_id"]) == {100, 200, 300} + class TestIndexValidation: def test_crossing_pair_without_cover_raises_at_index(self):