Validate scheduler models and cache paths

Makefile

@@ -52,7 +52,7 @@ check-dist:  ## check python sdist and wheel with check-dist
 check-types:  ## check python types with ty
 	ty check --python $$(which python)
 
-checks: check-dist
+checks: check-dist check-types
 
 # Alias
 check: checks

README.md

@@ -20,15 +20,19 @@ three questions:
 
 - **Should this be split, and by how much?** A closed form over each engine's
   measured throughput gives the share of rows that finishes both arms at the
-  same time. When the engines are comparable it approaches a 2x speedup; when
-  one is far faster than the other it correctly declines to bother.
+  same time. When the engines are comparable the predicted gain approaches
+  2x; when one is far faster than the other it shrinks toward 1x, and the
+  split reports that figure rather than hiding it. A device job's fixed cost
+  can rule a split out entirely, in which case the answer is host-only and
+  says why.
 - **Can this query survive being split?** Splitting rows means each engine
   computes a partial answer, and not every result can be reassembled from
   partials. Sums and counts add. A mean needs its sum and count carried
   separately rather than averaged. A top-k merges two k-lists. A grouped
   aggregation needs a key merge with the group straddling the boundary folded
-  back together. A distinct count cannot be combined at all without the
-  elements themselves. This library writes that taxonomy down and enforces it.
+  back together. An exact distinct count cannot be combined without the
+  elements themselves. This library writes that taxonomy down and refuses
+  any aggregate the caller has not declared mergeable under it.
 - **What if it can't?** Refusal is by name and default-deny. A result the merge
   step does not know how to combine is refused, explicitly and with a reason,
   rather than silently mis-merged into a wrong answer.
@@ -43,6 +47,16 @@ the machine it was measured on.
 The library is engine-agnostic. Both arms are injected as callables, and nothing
 here knows what is on the other end.
 
+## Install
+
+Python 3.12 or newer, and pydantic; nothing else.
+
+```bash
+pip install git+https://github.com/dau-dev/dau-scheduler
+```
+
+(The `dau-scheduler` name on PyPI is a placeholder until the first release.)
+
 ## Usage
 
 An arm is a callable. It is handed the query, the rows it owns, and the
@@ -92,23 +106,36 @@ execution.overlap_seconds  # how long both arms were in flight together
 be reassembled and it is refused before either engine starts:
 
 ```python
-execute_split(query, rows, split, merge, device=arm, host=arm, aggregates=[("average", "mean")])
-# SplitError: average: a mean is not row-partitionable; carry the sum and the
-# count as separate outputs and divide after the merge
+from dau_scheduler import SplitError
+
+try:
+    execute_split(query, rows, split, merge, device=arm, host=arm, aggregates=[("average", "mean")])
+except SplitError as refusal:
+    print(refusal)
+# average: a mean is not row-partitionable; carry the sum and the count as
+# separate outputs and divide after the merge
 ```
 
 ### Rows an arm never reads
 
 An engine running against data it already holds needs one number, not a list of
 rows: a row **count** if it holds its own copy, an **offset** if it holds a
 resident frame. `RowRange` is exactly that, and it costs two integers however
-many rows it names — so a share sweep over six million rows allocates nothing
-per call. An arm that does need the values asks for them and gets a typed
-refusal from a range rather than a table materialized behind the caller's back.
+many rows it names, so a share sweep over six million rows allocates nothing
+proportional to the row count. An arm that does need the values asks for them
+and gets a typed refusal from a range rather than a table materialized behind
+the caller's back.
 
 ```python
 from dau_scheduler import RowRange
 
+
+def resident_arm(query, rows, build):
+    # an engine that holds the table: it reads the offset and the length,
+    # never a row
+    return {"revenue": 0, "len": len(rows)}
+
+
 execute_split(query, RowRange(0, 6_000_000), split, merge, device=resident_arm, host=resident_arm)
 ```
 
@@ -121,17 +148,19 @@ price every later run from it.
 ```python
 from dau_scheduler import ProfileCache, plan_split, profile_from_spans
 
+# (start_us, end_us, node name) per plan node, as the engine reported them
+engine_timings = [(0, 11_200, "optimization"), (11_200, 12_900, "scan"), (12_900, 14_600, "filter")]
 profile = profile_from_spans(
-    engine_timings,  # (start_us, end_us, node name) per plan node
+    engine_timings,
     rows=1_000,
     plan_key="revenue-by-day",
     is_fixed=lambda node: node.startswith("optimization"),
     engine_version="1.42.1",
 )
-ProfileCache("~/profiles").put(profile)
+ProfileCache("profiles").put(profile)
 
 if profile.is_stale_for(6_000_000, engine_version="1.42.1"):
-    ...  # re-measure rather than extrapolate
+    pass  # re-measure rather than extrapolate across orders of magnitude
 plan_split("collaborative", device_seconds=0.0186, profile=profile, rows=6_000_000)
 ```
 

dau_scheduler/merge.py

@@ -15,9 +15,11 @@
   carry sum and count SEPARATELY and the division happens after the merge --
   which is a property of what the engines emit, not something a merge step
   can recover.
-* **A distinct count cannot be combined at all** without the elements it
-  counted. Two ranges that each saw 100 distinct keys may between them have
-  seen 100 or 200, and nothing in the partials says which.
+* **An exact distinct count cannot be combined** without the elements it
+  counted. Two ranges that each saw ``n`` distinct keys may between them have
+  seen anywhere from ``n`` to ``2n``, and nothing in the partials says
+  which. (A mergeable sketch would combine, as an approximation; it is not
+  what these engines emit, so it is not what this taxonomy admits.)
 * **Variance and standard deviation** merge only from count, sum and
   sum-of-squares carried separately, for the same reason as the mean.
 * **A median, or any order statistic of the whole**, needs the values rather
@@ -94,14 +96,14 @@
 # aggregate and the result shape it was asked of.
 AGGREGATE_REFUSALS: dict[str, str] = {
     "mean": "a mean is not row-partitionable; carry the sum and the count as separate outputs and divide after the merge",
-    "n_unique": "a distinct count is not row-partitionable: counts over disjoint ranges cannot be combined without the elements themselves",
+    "n_unique": "an exact distinct count is not row-partitionable: counts over disjoint ranges cannot be combined without the elements themselves",
     "median": "a median is not row-partitionable: an order statistic of the whole needs the values, not two partial answers",
     "std": "a standard deviation is not row-partitionable; carry count, sum and sum-of-squares and derive it after the merge",
     "var": "a variance is not row-partitionable; carry count, sum and sum-of-squares and derive it after the merge",
 }
 
 
-def _no_combine(partials: Sequence[Any]) -> Any:
+def _no_combine(_partials: Sequence[Any]) -> Any:
     raise SplitError("this merge rule carries no combine step, so its partials cannot be reassembled")
 
 

dau_scheduler/posture.py

@@ -4,11 +4,12 @@
 second engine runs. Everything else follows from it.
 
 **The postures optimize different things and are not quality tiers.** On a
-measured pairing where one engine alone finished in 27.6 ms and the other in
-18.6 ms, running it entirely on the second engine costs about 40% of the wall
-clock that splitting reaches -- and hands back the first engine entirely in
-exchange. A planner that only minimized wall time would pick wrongly for a
-caller whose processors have other work to do.
+measured pairing (the collaborative sweep of 2026-08-12 in the benchmarks
+ledger) the host alone took 27.6 ms and the device alone 18.6 ms; the split
+finished in 11.9 ms. Running everything on the device therefore takes 1.57x
+the wall clock of splitting, and in exchange hands the host back entirely.
+A planner that only minimized wall time would pick wrongly for a caller whose
+processors have other work to do.
 
 The closed form under ``collaborative`` is the balance point where both
 engines finish together::
@@ -20,7 +21,8 @@
 
 The speedup expression is also the reason to check whether a split is worth
 building at all: **collaborative placement pays most when the engines are
-comparable.** At parity it returns 2x. Against an engine ten times faster
+comparable.** From the form above (these are analytic values, not
+measurements): at parity it returns 2x; against an engine ten times faster
 than the host it buys 1.1x over simply handing the whole query over, which
 is unlikely to repay the machinery.
 
@@ -34,6 +36,7 @@
 
 from __future__ import annotations
 
+import math
 from typing import ClassVar, Literal, Protocol, runtime_checkable
 
 from pydantic import BaseModel, model_validator
@@ -62,21 +65,32 @@ def estimate(self, rows: int) -> float: ...
 class WorkSplit(BaseModel):
     """How one query's rows divide between the two engines."""
 
-    model_config: ClassVar = {"frozen": True}
+    model_config: ClassVar = {"frozen": True, "extra": "forbid"}
 
     posture: str
     device_share: float
     rationale: str
     predicted_speedup: float | None = None
 
-    @model_validator(mode="after")
-    def _share_is_a_fraction(self) -> WorkSplit:
-        if not 0.0 <= self.device_share <= 1.0:
+    def check(self) -> WorkSplit:
+        """Re-run the invariants and return ``self``.
+
+        Construction validates, but ``model_copy(update=...)`` does not, so
+        every consumer calls this before acting on a split rather than
+        trusting that the instance it was handed went through ``__init__``.
+        """
+        if not math.isfinite(self.device_share) or not 0.0 <= self.device_share <= 1.0:
             raise ValueError(f"device_share {self.device_share} is not a fraction")
         if self.posture not in POSTURES:
             raise ValueError(f"unknown posture {self.posture!r}; known: {POSTURES}")
+        if self.predicted_speedup is not None and (not math.isfinite(self.predicted_speedup) or self.predicted_speedup <= 0):
+            raise ValueError(f"predicted_speedup must be finite and positive, got {self.predicted_speedup}")
         return self
 
+    @model_validator(mode="after")
+    def _share_is_a_fraction(self) -> WorkSplit:
+        return self.check()
+
 
 def plan_split(
     posture: Posture = "collaborative",
@@ -108,7 +122,7 @@ def plan_split(
     ``device_fixed_seconds`` is what one device job costs before it has
     processed a row -- launching it, polling it, reading its records -- a
     per-invocation term the proportional form omitted and one measurement
-    put at a few milliseconds beside a 13 ms host time (2026-09-27). With it
+    put at a few milliseconds beside a 12.3 ms host time (2026-09-27). With it
     the balance is ``(1 - s) * cpu = fixed + s * device``, and a share that
     cannot beat the host alone (the fixed cost is already most of the host
     time) is a ``cpu`` split that says why, not a split predicted to win by
@@ -170,14 +184,15 @@ def plan_split(
         )
 
     if posture == "adaptive":
-        # A refusal, and not because it is unfinished: adaptive chooses a
-        # BOUNDARY in the plan rather than a row share, so it cannot be
-        # expressed as a WorkSplit and needs inputs this signature does not
-        # carry -- candidate cuts and a per-node cost. Answering it with a
-        # row split would silently substitute a different decision.
-        raise NotImplementedError(
+        # A refusal, not an unfinished branch: adaptive chooses a BOUNDARY
+        # in the plan rather than a row share, so it cannot be expressed as
+        # a WorkSplit and needs inputs this signature does not carry
+        # (candidate cuts and a per-node cost). The consumer that holds the
+        # plan answers it; answering it here with a row split would silently
+        # substitute a different decision.
+        raise PostureError(
             "adaptive is not a row split: it chooses where to cut the plan, which needs candidate cuts and a "
-            "profiled per-node cost rather than two whole-query figures"
+            "profiled per-node cost rather than two whole-query figures; plan it where the plan is"
         )
 
     raise PostureError(f"unknown posture {posture!r}; known: {POSTURES}")