[steps] Prevent Parallel from being in a Chain

CHANGELOG.md

@@ -6,6 +6,7 @@
 
 - `CacheDict`: deletions require a `write()` context (like writes). [#326]
 - `DumpContext.shared_file`: content suffixes must start with `.`. [#326]
+- `steps`: `Parallel` rejects input from an upstream step. [#328]
 
 [other]
 

exca/steps/helpers.py

@@ -128,7 +128,8 @@ class Parallel(Step):
     The variants run together under one shared backend, each caching under its
     own identity. ``run`` is for effect — read results back per variant via
     ``parallel.steps[k].lookup(value)``. It has no composable output (yields
-    ``None`` per input), so use it standalone, not as a non-terminal chain step.
+    ``None`` per input), so it cannot consume another step's output — run it
+    standalone.
 
     Example::
 
@@ -179,16 +180,18 @@ def _unify_infra(self) -> None:
                     f"steps; {self.infra!r} differs from {step.infra!r}"
                 )
 
-    def _uid_steps(self) -> list[Step]:
-        return []  # no identity of its own
-
     def lookup(self, *args: tp.Any, **kwargs: tp.Any) -> tp.NoReturn:
         raise TypeError(
             "Parallel has no cache of its own; look up a variant instead, "
             "e.g. parallel.steps[k].lookup(value)"
         )
 
     def _dispatch(self, batch: items.StepItems) -> items.StepItems:
+        if batch._upstream:
+            raise TypeError(
+                "Parallel has no output to pass on, so it cannot consume another "
+                "step's output; run it standalone."
+            )
         return self._run_items(batch)  # not infra._run(self): dispatch variants
 
     def _run_items(self, batch: items.StepItems) -> items.StepItems:

exca/steps/test_helpers.py

@@ -142,6 +142,11 @@ def test_run_caches_each_variant_under_own_identity(tmp_path: Path) -> None:
     assert [s.lookup(5.0).result() for s in sweep.steps] == [10.0, 15.0, 20.0]
     infra: tp.Any = {"backend": "Cached", "folder": tmp_path}
     assert conftest.Mult(coeff=3.0, infra=infra).lookup(5.0).cached()
+    Chain(steps=[_sweep(tmp_path)], infra=infra).run_many([1.0])  # allowed: no upstream
+    folders = conftest.extract_cache_folders(tmp_path)
+    assert any(f.startswith("type=Parallel") for f in folders), (
+        f"chain-leading Parallel must cache under its own identity, got {folders}"
+    )
 
 
 def test_generator_variants_no_items(tmp_path: Path) -> None:
@@ -163,6 +168,8 @@ def test_invalid_inputs_rejected(tmp_path: Path) -> None:
         Parallel(steps=conflicting, infra=infra)
     with pytest.raises(TypeError, match="parallel.steps"):
         _sweep(tmp_path).lookup(5.0)
+    with pytest.raises(TypeError, match="cannot consume another step's output"):
+        Chain(steps=[conftest.Add(value=100.0), _sweep(tmp_path)]).run_many([3.0])
 
 
 @pytest.mark.parametrize("folder_first", (True, False))