feat: optional step_id for custom output folder names

Allow readable step folders and xcf_stack layer labels via p.step(..., step_id=...)
while keeping the per-module auto counter unchanged.

Co-authored-by: Cursor <cursoragent@cursor.com>

tests/test_pipeline.py

@@ -159,6 +159,70 @@ class TestManifest:
         assert data["finished_at"] is not None
 
 
+class TestCustomStepId:
+    def test_custom_step_id_output_folder(self, input_dir: Path, output_base: Path) -> None:
+        @register
+        class NamedModule(BaseModule):
+            name = "named_tracker"
+
+            def run(self, ctx) -> None:
+                ctx.output_dir.mkdir(parents=True, exist_ok=True)
+                for src in ctx.input_paths:
+                    shutil.copy2(src, ctx.output_dir / src.name)
+
+        with Pipeline(name="named_test", input_dir=input_dir, output_base=output_base, verbose=False) as p:
+            ref = p.step("named_tracker", inputs="input", step_id="input_bokeh")
+            root = p.run()
+
+        assert ref.step_id == "input_bokeh"
+        assert ref.output_dir_name == "input_bokeh"
+        assert (root / "input_bokeh").is_dir()
+        assert not (root / "named_tracker_01").exists()
+
+        unregister("named_tracker")
+
+    def test_module_counter_independent_of_custom_step_id(
+        self, input_dir: Path, output_base: Path
+    ) -> None:
+        @register
+        class CounterModule(BaseModule):
+            name = "counter_tracker"
+
+            def run(self, ctx) -> None:
+                ctx.output_dir.mkdir(parents=True, exist_ok=True)
+                for src in ctx.input_paths:
+                    shutil.copy2(src, ctx.output_dir / src.name)
+
+        with Pipeline(name="counter_test", input_dir=input_dir, output_base=output_base, verbose=False) as p:
+            p.step("counter_tracker", inputs="input")
+            p.step("counter_tracker", inputs="input", step_id="custom_mid")
+            p.step("counter_tracker", inputs="input")
+            root = p.run()
+
+        assert (root / "counter_tracker_01").is_dir()
+        assert (root / "custom_mid").is_dir()
+        assert (root / "counter_tracker_03").is_dir()
+        assert not (root / "counter_tracker_02").exists()
+
+        unregister("counter_tracker")
+
+    def test_duplicate_step_id_rejected(self, input_dir: Path) -> None:
+        with Pipeline(name="dup_id", input_dir=input_dir, verbose=False) as p:
+            p.step("imagemagick_grayscale", inputs="input", step_id="my_step")
+            with pytest.raises(ValidationError, match="Duplicate step_id"):
+                p.step("imagemagick_grayscale", inputs="input", step_id="my_step")
+
+    def test_empty_step_id_rejected(self, input_dir: Path) -> None:
+        with Pipeline(name="empty_id", input_dir=input_dir, verbose=False) as p:
+            with pytest.raises(ValidationError, match="must not be empty"):
+                p.step("imagemagick_grayscale", inputs="input", step_id="")
+
+    def test_step_id_with_slash_rejected(self, input_dir: Path) -> None:
+        with Pipeline(name="slash_id", input_dir=input_dir, verbose=False) as p:
+            with pytest.raises(ValidationError, match="path separators"):
+                p.step("imagemagick_grayscale", inputs="input", step_id="bad/name")
+
+
 class TestPipelineValidation:
     def test_empty_pipeline_rejected(self, input_dir: Path) -> None:
         with Pipeline(name="empty", input_dir=input_dir, verbose=False) as p: