[Proto] Add a Fit step and FitCohort

CHANGELOG.md

@@ -2,6 +2,8 @@
 
 ## [Unreleased]
 
+- `steps`: `Fit` primitive — fit one artifact over a cohort of items and transform the items accordingly.
+
 ## 0.5.29 - 26-07-28
 
 - `ConfDict`: added `ConfDict` operations `ConfDict.ops.DELETE`, `ConfDict.ops.BEFORE`, and `ConfDict.ops.AFTER`; `ConfDict.ops.REPLACE` replaces `ConfDict.OVERRIDE`. [#310]

docs/internal/steps/caching.md

@@ -94,6 +94,12 @@ whole batch set, so a sweep (many step variants dispatched together) packs
 into one submitit array (one pool for pool backends). `_mark_recomputed`
 records force/retry batches per attempt.
 
+`Backend._grouped()` is how a sweep gets there: inside it, `_run` only
+prepares and queues its batch, and the group is claimed and executed once
+on exit. `Parallel` uses it so its variants still go through the regular
+`Step._dispatch` (step resolution, `Fit` cohort resolution) instead of
+driving the backend primitives itself.
+
 The session locks `inflight.db` only — direct user calls to
 `LookupHandle.clear_cache()` race against in-flight workers. Results are
 not round-tripped through the job pickle (would be wasteful under

docs/steps/items.md

@@ -133,6 +133,37 @@ at a time (never the full set in memory, never round-tripped through
 the job pickle). Execution order within a batch is non-deterministic;
 output order matches input order.
 
+## Fitting on the items
+
+A `Fit` step derives one artifact from many items, then transforms each item with it — normalization statistics, a PCA basis, a trained model. `fit_many` fits; `run_many` transforms with what is already fitted:
+
+```python
+class Normalize(steps.Fit):
+    def _fit(self, values):        # the cohort
+        stacked = np.concatenate(list(values))
+        return stacked.mean(0), stacked.std(0)
+
+    def _run(self, value):         # one item
+        mean, std = self.fitted
+        return (value - mean) / std
+
+
+norm = Normalize(infra={"backend": "Cached", "folder": cache})
+for value in norm.fit_many(train_paths):
+    train(value)                   # fitted on this cohort, then transformed
+for value in norm.run_many(test_paths):
+    evaluate(value)                # same artifact, novel items
+```
+
+`_fit` receives the cohort as an iterable it can stream, and iterate
+again (one upstream read per pass).
+
+The cohort's identity — the fingerprint of its items, or the name passed to `fit_many(..., cohort="train")` — settles before anything runs, and the step that runs is a copy carrying it, so the artifact and every downstream cache are scoped to it. Bind an existing named cohort with `fit_many(cohort="train")`. Another cohort or upstream takes another config, so `clone()` it.
+
+The fit runs where the step is dispatched from, ahead of any split,
+and is cached under `infra`. A `Fit` under a backend that shards the
+cohort raises rather than fitting on a shard.
+
 ## What's stable
 
 Pinned by tests — safe to rely on:

docs/steps/reference.rst

@@ -11,7 +11,7 @@ Core classes
 ------------
 
 .. autoclass:: exca.steps.Step
-    :members: run, run_many, lookup, clone, item_uid, CACHE_TYPE
+    :members: run, run_many, fit_many, lookup, clone, item_uid, CACHE_TYPE
 
 .. autoclass:: exca.steps.Chain
     :show-inheritance:
@@ -90,3 +90,12 @@ Helpers
 
 .. autoclass:: exca.steps.helpers.Func
     :members:
+
+
+Fitting over items
+------------------
+
+.. autoclass:: exca.steps.Fit
+    :show-inheritance:
+    :members: fitted, ARTIFACT_CACHE_TYPE
+    :private-members: _fit, _cohort_uids

exca/cachedict/handlers.py

@@ -383,9 +383,8 @@ def _dump_value(cls, ctx: DumpContext, val: tp.Any, key: str) -> tp.Any:
                         f"Found #type={val['#type']!r} which is not a registered handler."
                     )
             return {k: cls._dump_value(ctx, v, f"{key}[{k}]") for k, v in val.items()}
-        if isinstance(val, (list, tuple)):
-            items = [cls._dump_value(ctx, v, f"{key}[{i}]") for i, v in enumerate(val)]
-            return tuple(items) if isinstance(val, tuple) else items
+        if isinstance(val, (list, tuple)):  # as a list: `_load_value` walks lists only
+            return [cls._dump_value(ctx, v, f"{key}[{i}]") for i, v in enumerate(val)]
         handler = DumpContext._find_handler(type(val))
         if handler is not None or hasattr(val, "__dump_info__"):
             ctx.key = key
@@ -416,7 +415,7 @@ def _delete_value(cls, ctx: DumpContext, val: tp.Any) -> None:
             else:
                 for v in val.values():
                     cls._delete_value(ctx, v)
-        elif isinstance(val, list):
+        elif isinstance(val, (list, tuple)):  # tuple: only in legacy entries
             for item in val:
                 cls._delete_value(ctx, item)
 
@@ -426,7 +425,7 @@ def _load_value(cls, ctx: DumpContext, val: tp.Any) -> tp.Any:
             if "#type" in val:
                 return ctx.load(val)
             return {k: cls._load_value(ctx, v) for k, v in val.items()}
-        if isinstance(val, list):
+        if isinstance(val, (list, tuple)):  # tuple: only in legacy entries
             return [cls._load_value(ctx, item) for item in val]
         return val
 

exca/cachedict/test_dumpcontext.py

@@ -300,6 +300,19 @@ def tracking_dump(
     np.testing.assert_array_almost_equal(loaded["weights"], [[1.0, 2.0], [3.0, 4.0]])
 
 
+def test_auto_tuple_loads_as_list(tmp_path: Path) -> None:
+    ctx = DumpContext(tmp_path, key="test")
+    with ctx:
+        info = ctx.dump((np.array([1.0, 2.0]), 3), cache_type="Auto")
+    loaded = ctx.load(info)
+    assert isinstance(loaded, list), "json storage cannot hold a tuple either"
+    np.testing.assert_array_almost_equal(loaded[0], [1.0, 2.0])
+    assert loaded[1] == 3
+
+    legacy = ctx.load({**info, "content": tuple(info["content"])})  # tuple: old entry
+    np.testing.assert_array_almost_equal(legacy[0], [1.0, 2.0])
+
+
 class _Opaque:
     """Module-level class so it's picklable (local classes are not)."""
 

exca/steps/__init__.py

@@ -29,3 +29,4 @@
 from . import helpers as helpers
 from .base import Chain as Chain
 from .base import Step as Step
+from .fit import Fit as Fit