refactor[next]: replace workflow combinators with explicit typed pipelines

docs/development/ADRs/next/0029-Backend-Structure-and-Pipeline-Simplification.md

@@ -122,6 +122,27 @@ methods and callable steps (`Step[S, T] = Callable[[S], T]`). The standard compi
 | Steps compose across backends               | `Step[S, T]` is a `Callable` — every existing step already satisfies it |
 | Linear workflows                            | unchanged (`Transforms` keeps its input-dependent step *selection*)     |
 
+Static checking covers the stage types a step consumes and produces, not the language of their
+code: the `CompilePipeline` fields use unparameterized `ProgramSource` / `ExtensionSource`, so mypy
+accepts a pipeline mixing C++ and SDFG steps. Parameterizing `CompilePipeline` over the code spec
+only helps once the builders return parameterized pipelines, so it is left to a follow-up.
+
+### Cross-step agreement is checked on construction
+
+The builders of [0028](0028-Plain-Builders-Instead-of-Factories.md) keep default and partially
+customized steps in sync, but fully custom steps and `dataclasses.replace` can assemble a pipeline
+or a backend without any builder. The assembling objects therefore check device agreement
+themselves, in `__post_init__`, which `dataclasses.replace` re-runs:
+
+- `CompilePipeline`: all steps that declare a device (the `backend.DeviceConfigurable` protocol,
+  looking through `CachedStep`) declare the same one, exposed as `CompilePipeline.device_type`.
+- `CompiledBackend`: the allocator's `__gt_device_type__` matches the device the `compiler`
+  declares.
+
+The checks never mutate and only see components that declare a device; a custom step without
+`device_type` is not checked. Settings shared by the steps of one backend family, such as the DaCe
+external workspace and its `transient_memory_mode`, are that family's responsibility.
+
 ### Stage observability
 
 In the target design, a `stage_hook(name, artifact)` event hook on `instrumentation.hook_machinery` is
@@ -198,7 +219,8 @@ The overridable `compile`/`load_artifact` methods, `DaCeBackend` subclass, opaqu
 compile” path and `CompilationTask.no_offload_reason` are removed. The pipeline PR removes the
 combinator framework (both mixins, `NamedStepSequence`, `MultiWorkflow`, `StepSequence`,
 `make_step`, `.chain` and the three adapters), and dissolves `otf.definitions`, `otf.code_specs`,
-`otf.recipes` and `otf.toolchain`. `CachedStep` remains unchanged.
+`otf.recipes` and `otf.toolchain`. `CachedStep` is kept, minus the mixin-provided `.replace` /
+`.chain`.
 
 ### Cache-key rotation
 
@@ -219,10 +241,11 @@ this argues against spreading it across releases. Backend restructuring itself r
 ### Phasing and earlier decisions
 
 #2741 lands the backend structure, loader, naming except `CompilePipeline`, and removal of the
-opaque compile path. #2742 adds stage observability, including `Toolchain.translate`; #2743
-replaces the combinators and renames `OTFCompileWorkflow` to `CompilePipeline`. Thus the pipeline
-and observability sections describe the target design, not APIs already available at #2741.
-Embedded backends follow separately under [ADR 0030](0030-Embedded-Backends.md) (#2956).
+opaque compile path. Embedded backends follow, decided separately in
+[ADR 0030](0030-Embedded-Backends.md) (#2956). #2742 adds stage observability, including
+`Toolchain.translate`; #2743 replaces the combinators, renames `OTFCompileWorkflow` to
+`CompilePipeline` and adds the construction-time device checks. Thus the pipeline, agreement and
+observability sections describe the target design, not APIs already available at #2741.
 
 This partially supersedes [0011 - On The Fly Compilation](0011-On_The_Fly_Compilation.md) (the
 workflow-combinator framework and the `otf.step_types` naming; its stage vocabulary and

docs/user/next/advanced/HackTheToolchain.md

@@ -4,8 +4,6 @@ import typing
 
 from gt4py import next as gtx
 from gt4py.next.otf import workflow
-from gt4py.next.ffront import field_operator_ast as foast, stages as ff_stages
-from gt4py import eve
 ```
 
 <link href="https://fonts.googleapis.com/icon?family=Material+Icons" rel="stylesheet"><script src="https://spcl.github.io/dace/webclient2/dist/sdfv.js"></script>
@@ -16,44 +14,31 @@ from gt4py import eve
 > sections) is global and not a user customization point: a backend does not carry a frontend
 > (ADR 0029). A modified `Transforms` can only be used through the internal
 > `gtx.backend.Toolchain`, e.g. in gt4py's own tests. What a backend does is customized through
-> its `compiler` and `loader`, see "Alternative Workflow" below.
+> its `compiler` and `loader`, see "Alternative Pipeline" below.
 
 ## Replace Steps
 
-```python
-cached_lowering_toolchain = gtx.backend.DEFAULT_FRONTEND.replace(
-    past_to_itir=gtx.ffront.past_to_itir.past_to_gtir_factory(cached=False)
-)
-```
-
-## Skip Steps / Change Order
+Pipelines are frozen dataclasses whose fields are the steps, so a variant is built with `dataclasses.replace`.
 
 ```python
-DUMMY_FOP = workflow.ProgramWithArgs(
-    definition=ff_stages.DSLFieldOperatorDef(definition=None), args=None
+cached_lowering_toolchain = dataclasses.replace(
+    gtx.backend.DEFAULT_FRONTEND,
+    past_to_itir=gtx.ffront.past_to_itir.past_to_gtir_factory(cached=False),
 )
 ```
 
-```python
-gtx.backend.DEFAULT_FRONTEND.step_order(DUMMY_FOP)
-```
-
-```python
-@dataclasses.dataclass(frozen=True)
-class SkipLinting(gtx.backend.Transforms):
-    def step_order(self, inp):
-        order = super().step_order(inp)
-        if "past_lint" in order:
-            order.remove("past_lint")  # not running "past_lint"
-        return order
+## Skip Steps
 
+Which steps run is decided by `Transforms.__call__` from the type of the input definition; that selection is not configurable. Behavior is customized by replacing a step, so skipping one means replacing it with an identity step.
 
-same_steps = dataclasses.asdict(gtx.backend.DEFAULT_FRONTEND)
-skip_linting_transforms = SkipLinting(**same_steps)
-skip_linting_transforms.step_order(DUMMY_FOP)
+```python
+skip_linting_transforms = dataclasses.replace(
+    gtx.backend.DEFAULT_FRONTEND,
+    past_lint=lambda past_def: past_def,  # identity step: linting skipped
+)
 ```
 
-## Alternative Workflow
+## Alternative Pipeline
 
 A compiled backend is built from one configuration, `GTFNConfig` (or `DaCeConfig`),
 which holds the settings all steps must agree on: the device, the build type,
@@ -86,14 +71,43 @@ class MyCodeGen: ...
 class Cpp2BindingsGen: ...
 
 
-pure_cpp2_workflow = gtfn.make_gtfn_compile_workflow(
+pure_cpp2_pipeline = gtfn.make_gtfn_compile_workflow(
     gtfn.GTFNConfig(cmake_build_type=gtx.config.CMakeBuildType.DEBUG, cached_translation=False),
     translation=lambda cfg: MyCodeGen(),
     bindings=lambda cfg: Cpp2BindingsGen(),
 )
 ```
 
-## Invent new Workflow Types
+An existing pipeline, such as the `compiler` of a pre-built backend, can also be varied after the fact with `dataclasses.replace`. Unlike a step builder, a step passed this way is used exactly as given, with two consequences:
+
+- **Caching is not carried over.** The builders wrap the translation step in a persistent `CachedStep`. Replacing `translation` replaces that wrapper as well, so the variant does not cache translations unless you wrap the new step yourself. Passing a step builder instead keeps the caching.
+- **Device agreement is checked, but only for steps that declare a device.** If a step has a `device_type` attribute, it must match every other step that declares one, and the backend's allocator. `CompilePipeline` and `CompiledBackend` check this whenever they are constructed, `dataclasses.replace` included, and raise a `ValueError` on a mismatch. Steps without a `device_type`, like `MyCodeGen` above, are not checked, so keeping them consistent is up to you.
+
+```python
+from gt4py.next import fingerprinting
+
+gtfn_pipeline = gtx.program_processors.runners.gtfn.run_gtfn.compiler
+cached_translation = gtfn_pipeline.translation  # a `CachedStep` around the translation step
+no_transforms = dataclasses.replace(cached_translation.step, enable_itir_transforms=False)
+
+uncached_variant = dataclasses.replace(gtfn_pipeline, translation=no_transforms)
+cached_variant = dataclasses.replace(
+    gtfn_pipeline,
+    translation=workflow.CachedStep.in_memory(
+        step=no_transforms, input_fingerprinter=fingerprinting.strict_fingerprinter
+    ),
+)
+
+gpu_translation = dataclasses.replace(cached_translation.step, device_type=gtx.DeviceType.CUDA)
+try:
+    dataclasses.replace(gtfn_pipeline, translation=gpu_translation)
+except ValueError as error:
+    print(error)
+```
+
+## Invent new Pipeline Types
+
+A pipeline is just a frozen dataclass of steps with an explicit, fully typed `__call__`. Nothing else is needed, so a non-linear shape is written the same way as a linear one.
 
 ```mermaid
 graph LR
@@ -112,15 +126,11 @@ OUT_T = typing.TypeVar("OUT_T")
 
 
 @dataclasses.dataclass(frozen=True)
-class FullyModularDiamond(
-    workflow.ChainableWorkflowMixin[IN_T, OUT_T],
-    workflow.ReplaceEnabledWorkflowMixin[IN_T, OUT_T],
-    typing.Protocol[IN_T, OUT_T, A_T, B_T, X_T, Y_T],
-):
-    split: workflow.Workflow[IN_T, tuple[A_T, X_T]]
-    track_a: workflow.Workflow[A_T, B_T]
-    track_x: workflow.Workflow[X_T, Y_T]
-    combine: workflow.Workflow[tuple[B_T, Y_T], OUT_T]
+class Diamond(typing.Generic[IN_T, OUT_T, A_T, B_T, X_T, Y_T]):
+    split: workflow.Step[IN_T, tuple[A_T, X_T]]
+    track_a: workflow.Step[A_T, B_T]
+    track_x: workflow.Step[X_T, Y_T]
+    combine: workflow.Step[tuple[B_T, Y_T], OUT_T]
 
     def __call__(self, inp: IN_T) -> OUT_T:
         a, x = self.split(inp)
@@ -129,20 +139,10 @@ class FullyModularDiamond(
         return self.combine((b, y))
 
 
-@dataclasses.dataclass(frozen=True)
-class PartiallyModularDiamond(
-    workflow.ChainableWorkflowMixin[IN_T, OUT_T],
-    workflow.ReplaceEnabledWorkflowMixin[IN_T, OUT_T],
-    typing.Protocol[IN_T, OUT_T, A_T, B_T, X_T, Y_T],
-):
-    track_a: workflow.Workflow[A_T, B_T]
-    track_x: workflow.Workflow[X_T, Y_T]
-
-    def split(inp: IN_T) -> tuple[A_T, X_T]: ...
-
-    def combine(b: B_T, y: Y_T) -> OUT_T: ...
-
-    def __call__(inp: IN_T) -> OUT_T:
-        a, x = self.split(inp)
-        return self.combine(b=self.track_a(a), y=self.track_x(x))
+Diamond(
+    split=lambda inp: (inp, inp),
+    track_a=lambda a: a + 1,
+    track_x=lambda x: x * 2,
+    combine=lambda by: by[0] + by[1],
+)(3)
 ```