Repository: GridTools/gt4py
Pull request #1224 (2023-03-28 12:31:24)

refactor[next]: interface for creating workflow variants.
Added:
- `workflow.NamedStepSequence`: 
    baseclass for workflows consisting of a sequence of named (and type-hinted) steps 
    (the names can be used with workflow.replace)

- `workflow.replace`: 
    currently a proxy to `dataclasses.replace`, used to build variants of existing workflow instances
    
- `program_processors.otf_compile_executor`
  - `OTFCompileExecutor`:
      `ProgramExecutor` wrapper around any `OTFCompileExecutor` or `CachedStep` (with `OTFCompileExecutor` inside)
      
- `workflow.recipes`
  - `OTFCompileWorkflow`:
      A `NamedStepSequence` composing the typical steps: translation -> bindings -> compilation -> decoration

Changed:
- `GTFNExecutor` replaced by more generic `OTFCompileExecutor`:
    The workflow is now exposed and can be used to create variants through `workflow.replace`

## Description

## Requirements

- [X] All fixes and/or new features come with corresponding tests.
- [x] Important design decisions have been documented in the approriate ADR inside the [docs/development/ADRs/](docs/development/ADRs/Index.md) folder.



Patch:
diff --git a/docs/development/ADRs/0011-On_The_Fly_Compilation.md b/docs/development/ADRs/0011-On_The_Fly_Compilation.md
index 4768c8c8..14f61bf0 100644
--- a/docs/development/ADRs/0011-On_The_Fly_Compilation.md
+++ b/docs/development/ADRs/0011-On_The_Fly_Compilation.md
@@ -7,7 +7,7 @@ tags: [backend, bindings, build, compile, otf]
 - **Status**: valid
 - **Authors**: Rico Häuselmann (@DropD)
 - **Created**: 2022-09-12
-- **Updated**: 2022-10-20
+- **Updated**: 2023-04-03
 
 This supersedes [0009 - Compiled Backend Integration](0009-Compiled_Backend_Integration.md) and concentrates on the API design for on-the-fly compilation of GT4Py programs and all the steps in between IR and compiled Python extension.
 
@@ -155,33 +155,82 @@ There is a good use case that justifies the extra effort. Such as implementing c
 ### Statically Typed Workflows
 
 - decided: 2022-09-13
-- revised: --
-
-Workflow steps are generic on input and output type. Workflows are generic on input, intermediary and output type. This allows the use of static type checkers to ensure that the composed workflow has the intended input and output types, as well as that it is passed the correct argument types and that it's return value is used correctly.
+- revised: 2023-03-29
 
-In order to achieve this, workflows are implemented as nested pairs (or more generally, finite sized tuples) of steps, with the output type of the first step matching the input type of the second step in the pair. More complex workflow components than pairs might be introduced if the "Linear Workflows" decision is revised.
+Workflow steps are generic on input and output type and the chaining function is generic on the input, intermediary and (new) output type. This allows the use of static type checkers to ensure that the composed workflow has the intended input and output types, as well as that it is passed the correct argument types and that it's return value is used correctly.
 
-To allow building workflows in a linear-looking way in code, the first step can be wrapped in a `otf.workflow.Step` instance, on which `.chain` can be called with the next step. The result will be a workflow instance with `.chain` available also.
+To allow building workflows in a linear-looking way in code, the `otf.workflow.ChainableWorkflowMixin` mixin is provided.
+The first step can be wrapped in any chainable step type on which `.chain` can then be called with the next step. The result will be a workflow instance with `.chain` available also.
 
 ```
-workflow = <@todo put class name here too>(step1).chain(step2).chain(step3).chain(step4)
-# results in
-CombinedStep(
-  first=CombinedStep(
-    first=CombinedStep(
-      first=<@todo class name>(step1),
-      second=step2
-    ),
-    second=step3
-  ),
-  second=step4
-)
+step1: Workflow[I, A]
+step2: Workflow[A, B]
+step3: Workflow[B, C]
+step4: Workflow[C, O]
+workflow: Workflow[I, O] = StepSequence.start(step1).chain(step2).chain(step3).chain(step4)
 ```
 
+`otf.workflow.StepSequence` is only one example of a chainable step type.
+
+_edit 2023-03-29_
+
+`otf.workflow.NamedStepSequence` provides another way in which workflow steps can be composed, while keeping static typing somewhat intact.
+
+```python
+@dataclasses.dataclass(frozen=True)
+class ExampleWorkflow(NamedStepSequence):
+    parsing: Workflow[A, B]
+    lowering: Workflow[B, C]
+    first_pass: Workflow[C, C]
+    second_pass: Workflow[C, C]
+```
+
+As can be seen in the above example, the step types are statically hinted and mypy will catch wrong types on constructor calls. However, making sure that the steps are compatible in the order specified is up to the implementer of `ExampleWorkflow`.
+
 #### Reason
 
 It is in line with the rest of GT4Py to make static type checking for it's code objects as useful as possible. Any linear (and with extensions, non-linear) workflow can be represented in this way.
 
+_edit 2023-03-29_
+
+`NamedStepSequence` solves the following problem:
+
+By making it possible to quickly create a variant of an existing workflow (using `workflow.replace`), they make it unneccessary to escalate sub-workflow configuration.
+
+Coming back to the above example of `ExampleWorkflow`, instead of:
+
+```python
+examplewf_foo = Parser().chain(Lowerer(lower_flag=True)).chain(...)
+examplewf_bar = Parser().chain(Lowerer(lower_flag=False)).chain(...)
+```
+
+which avoids escalation, but is tedious, one might be tempted to implement a pattern like this:
+
+```python
+def make_workflow(lower_flag):
+    return Parser().chain(Lowerer(lower_flag=lower_flag)).chain(...)
+```
+
+which hardcodes the step classes and requires all sub-workflow configuration that needs to be accessible to be collected up front (escalated). One might fix the hardcoding of steps but the problem of escalated configuration just gets worse: now all configuration of all potentially allowed step types need to be escalated. `workflow.replace` in combination with `NamedStepSequence` solves both:
+
+```python
+examplewf_foo = ExampleWorkflow(
+    parsing=Parser(),
+    lowering=Lowerer(lower_flag=True),
+    ...
+)
+
+examplewf_bar = workflow.replace(
+    examplewf_foo,
+    lowering=dataclasses.replace(  # assuming the lowerer is a dataclass
+        examplewf_foo.lowering,
+        lower_flag=False
+    )
+)
+```
+
+This may be verbose but is much less dense semantically and much less error prone. A more terse interface may be added on top if warranted by usecases. This is of course a compromise when it comes to static typing. `workflow.replace` will take any argument which fulfills the `workflow.Workflow` protocol. This is a conscious tradeoff between usefulness of static typing and convenience of interface.
+
 #### Revise If
 
 There has to be a very strong reason to drop static type checking for composition and / or the resulting workflow. The author of this document can not imagine one at present.
diff --git a/docs/development/ADRs/0012-GridTools_Cpp_OTF_Steps.md b/docs/development/ADRs/0012-GridTools_Cpp_OTF_Steps.md
index a8cf594b..a256ee09 100644
--- a/docs/development/ADRs/0012-GridTools_Cpp_OTF_Steps.md
+++ b/docs/development/ADRs/0012-GridTools_Cpp_OTF_Steps.md
@@ -41,9 +41,7 @@ It is an instance of `program_processors.codegens.gtfn_modules.GTFNTranslationSt
 
 #### Executor
 
-`program_processors.runners.gtfn_cpu.run_gtfn` is the default `program_processors.processor_interface.ProgramExecutor` implementation of the GTFN backend. It is an instance of `program_processors.runners.gtfn_cpu.GTFNExecutor`, with sensible defaults for the configuration options allowed by that class. These include language settings and which builder should be used. BuildSystemProject configuration can be achieved through passing a fully configured builder factory.
-
-When called the `GTFNExecutor` instance composes a workflow from the above steps, the configured builder, importer and an arguments conversion decorator. The composed workflow is then called just in time for execution.
+`program_processors.runners.gtfn_cpu.run_gtfn` is the default `program_processors.processor_interface.ProgramExecutor` implementation of the GTFN backend. It sets sensible defaults for the configuration options on the individual steps. These include language settings and which builder should be used. Substep configuration can be achieved through passing one or more step instance to `dataclasses.replace(workflow.replace(run_gtfn.workflow, <name>=<instance>))`. Step instances can be obtained by replacing in a similar fashion.
 
 ## Alternatives Considered
 
diff --git a/src/gt4py/next/otf/compilation/compiler.py b/src/gt4py/next/otf/compilation/compiler.py
index df5e5124..32c54693 100644
--- a/src/gt4py/next/otf/compilation/compiler.py
+++ b/src/gt4py/next/otf/compilation/compiler.py
@@ -47,7 +47,15 @@ class BuildSystemProjectGenerator(Protocol[SrcL, LS, TgtL]):
 
 @dataclasses.dataclass(frozen=True)
 class Compiler(
-    step_types.CompilationStep[SourceLanguageType, LanguageSettingsType, languages.Python]
+    workflow.ChainableWorkflowMixin[
+        stages.CompilableSource[SourceLanguageType, LanguageSettingsType, languages.Python],
+        stages.CompiledProgram,
+    ],
+    workflow.ReplaceEnabledWorkflowMixin[
+        stages.CompilableSource[SourceLanguageType, LanguageSettingsType, languages.Python],
+        stages.CompiledProgram,
+    ],
+    step_types.CompilationStep[SourceLanguageType, LanguageSettingsType, languages.Python],
 ):
     """Use any build system (via configured factory) to compile a GT4Py program to a ``gt4py.next.otf.stages.CompiledProgram``."""
 
@@ -79,15 +87,6 @@ class Compiler(
             importer.import_from_path(src_dir / new_data.module), new_data.entry_point_name
         )
 
-    def chain(
-        self, step: workflow.Workflow[stages.CompiledProgram, T]
-    ) -> workflow.CombinedStep[
-        stages.CompilableSource[SourceLanguageType, LanguageSettingsType, languages.Python],
-        stages.CompiledProgram,
-        T,
-    ]:
-        return workflow.CombinedStep(first=self, second=step)
-
 
 class CompilationError(RuntimeError):
     ...
diff --git a/src/gt4py/next/otf/recipes.py b/src/gt4py/next/otf/recipes.py
new file mode 100644
index 00000000..d1445337
--- /dev/null
+++ b/src/gt4py/next/otf/recipes.py
@@ -0,0 +1,40 @@
+# GT4Py - GridTools Framework
+#
+# Copyright (c) 2014-2023, ETH Zurich
+# All rights reserved.
+#
+# This file is part of the GT4Py project and the GridTools framework.
+# GT4Py is free software: you can redistribute it and/or modify it under
+# the terms of the GNU General Public License as published by the
+# Free Software Foundation, either version 3 of the License, or any later
+# version. See the LICENSE.txt file at the top-level directory of this
+# distribution for a copy of the license or check <https://www.gnu.org/licenses/>.
+#
+# SPDX-License-Identifier: GPL-3.0-or-later
+from __future__ import annotations
+
+import dataclasses
+from typing import Generic, TypeVar
+
+from gt4py.next.otf import languages, stages, step_types, workflow
+
+
+SrcL = TypeVar("SrcL", bound=languages.LanguageTag)
+TgtL = TypeVar("TgtL", bound=languages.LanguageTag)
+LS = TypeVar("LS", bound=languages.LanguageSettings)
+
+
+@dataclasses.dataclass(frozen=True)
+class OTFCompileWorkflow(workflow.NamedStepSequence, Generic[SrcL, LS, TgtL]):
+    """The typical compiled backend steps composed into a workflow."""
+
+    translation: step_types.TranslationStep[SrcL, LS]
+    bindings: workflow.Workflow[
+        stages.ProgramSource[SrcL, LS],
+        stages.CompilableSource[SrcL, LS, TgtL],
+    ]
+    compilation: workflow.Workflow[
+        stages.CompilableSource[SrcL, LS, TgtL],
+        stages.CompiledProgram,
+    ]
+    decoration: workflow.Workflow[stages.CompiledProgram, stages.CompiledProgram]
diff --git a/src/gt4py/next/otf/step_types.py b/src/gt4py/next/otf/step_types.py
index 49d7e380..54fe2e53 100644
--- a/src/gt4py/next/otf/step_types.py
+++ b/src/gt4py/next/otf/step_types.py
@@ -16,7 +16,7 @@ from __future__ import annotations
 
 from typing import Protocol, TypeVar
 
-from gt4py.next.otf import languages, stages
+from gt4py.next.otf import languages, stages, workflow
 
 
 SrcL = TypeVar("SrcL", bound=languages.LanguageTag)
@@ -27,11 +27,13 @@ TgtL_co = TypeVar("TgtL_co", bound=languages.LanguageTag, covariant=True)
 LS_co = TypeVar("LS_co", bound=languages.LanguageSettings, covariant=True)
 
 
-class TranslationStep(Protocol[SrcL, LS]):
+class TranslationStep(
+    workflow.ReplaceEnabledWorkflowMixin[stages.ProgramCall, stages.ProgramSource[SrcL, LS]],
+    Protocol[SrcL, LS],
+):
     """Translate a GT4Py program to source code (ProgramCall -> ProgramSource)."""
 
-    def __call__(self, program_call: stages.ProgramCall) -> stages.ProgramSource[SrcL, LS]:
-        ...
+    ...
 
 
 class BindingStep(Protocol[SrcL, LS, TgtL]):
diff --git a/src/gt4py/next/otf/workflow.py b/src/gt4py/next/otf/workflow.py
index 9045e6a7..bde64287 100644
--- a/src/gt4py/next/otf/workflow.py
+++ b/src/gt4py/next/otf/workflow.py
@@ -15,7 +15,11 @@
 from __future__ import annotations
 
 import dataclasses
-from typing import Callable, Generic, Protocol, TypeVar
+import functools
+import typing
+from typing import Any, Callable, Generic, Protocol, TypeVar
+
+from typing_extensions import Self
 
 
 StartT = TypeVar("StartT")
@@ -27,7 +31,7 @@ IntermediateT = TypeVar("IntermediateT")
 HashT = TypeVar("HashT")
 
 
-def make_step(function: Workflow[StartT, EndT]) -> Step[StartT, EndT]:
+def make_step(function: Workflow[StartT, EndT]) -> ChainableWorkflowMixin[StartT, EndT]:
     """
     Wrap a function in the workflow step convenience wrapper.
 
@@ -44,9 +48,10 @@ def make_step(function: Workflow[StartT, EndT]) -> Step[StartT, EndT]:
     >>> times_two.chain(stringify)(3)
     '6'
     """
-    return Step(function)
+    return StepSequence.start(function)
 
 
+@typing.runtime_checkable
 class Workflow(Protocol[StartT_contra, EndT_co]):
     """
     Workflow protocol.
@@ -60,38 +65,104 @@ class Workflow(Protocol[StartT_contra, EndT_co]):
         ...
 
 
-@dataclasses.dataclass(frozen=True)
-class Step(Generic[StartT, EndT]):
+class ReplaceEnabledWorkflowMixin(Workflow[StartT_contra, EndT_co], Protocol):
     """
-    Workflow step convenience wrapper.
+    Subworkflow replacement mixin.
+
+    Any subclass MUST be a dataclass for `.replace` to work
+    """
+
+    def replace(self, **kwargs: Any) -> Self:
+        """
+        Build a new instance with replaced substeps.
 
-    Can wrap any callable which implements the workflow step protocol,
-    adds the .chain(other_step) method for convenience.
+        Raises:
+            TypeError: If `self` is not a dataclass.
+        """
+        if not dataclasses.is_dataclass(self.__class__):
+            raise TypeError(f"{self.__class__} is not a dataclass")
+        return dataclasses.replace(self, **kwargs)
+
+
+class ChainableWorkflowMixin(Workflow[StartT, EndT]):
+    def chain(self, next_step: Workflow[EndT, NewEndT]) -> ChainableWorkflowMixin[StartT, NewEndT]:
+        return make_step(self).chain(next_step)
+
+
+@dataclasses.dataclass(frozen=True)
+class NamedStepSequence(
+    ChainableWorkflowMixin[StartT, EndT],
+    ReplaceEnabledWorkflowMixin[StartT, EndT],
+):
+    """
+    Workflow with linear succession of named steps.
 
     Examples:
     ---------
-    >>> def times_two(x: int) -> int:
-    ...    return x * 2
+    >>> import dataclasses
 
-    >>> def stringify(x: int) -> str:
+    >>> def parse(x: str) -> int:
+    ...    return int(x)
+
+    >>> def plus_half(x: int) -> float:
+    ...    return x + 0.5
+
+    >>> def stringify(x: float) -> str:
     ...    return str(x)
 
-    >>> # create a workflow int -> int -> str
-    >>> Step(times_two).chain(stringify)(3)
-    '6'
-    """
+    >>> @dataclasses.dataclass(frozen=True)
+    ... class ParseOpPrint(NamedStepSequence[str, str]):
+    ...    parse: Workflow[str, int]
+    ...    op: Workflow[int, float]
+    ...    print: Workflow[float, str]
 
-    step: Workflow[StartT, EndT]
+    >>> pop = ParseOpPrint(
+    ...    parse=parse,
+    ...    op=plus_half,
+    ...    print=stringify
+    ... )
 
-    def __call__(self, inp: StartT) -> EndT:
-        return self.step(inp)
+    >>> pop.step_order
+    ['parse', 'op', 'print']
+
+    >>> pop(73)
+    '73.5'
 
-    def chain(self, step: Workflow[EndT, NewEndT]) -> CombinedStep[StartT, EndT, NewEndT]:
-        return CombinedStep(first=self.step, second=step)
+    >>> def plus_tenth(x: int) -> float:
+    ...   return x + 0.1
+
+
+    >>> pop.replace(op=plus_tenth)(73)
+    '73.1'
+    """
+
+    def __call__(self, inp: StartT) -> EndT:
+        """Compose the steps in the order defined in the `.step_order` class attribute."""
+        step_result: Any = inp
+        for step_name in self.step_order:
+            step_result = getattr(self, step_name)(step_result)
+        return step_result
+
+    @functools.cached_property
+    def step_order(self) -> list[str]:
+        """
+        Read step order from class definition by default.
+
+        Only attributes who are type hinted to be of a type that
+        conforms to the Workflow protocol are considered steps.
+        """
+        step_names: list[str] = []
+        annotations = typing.get_type_hints(self.__class__)
+        for field in dataclasses.fields(self):
+            field_type = annotations[field.name]
+            field_type = typing.get_origin(field_type) or field_type
+            if issubclass(field_type, Workflow):
+                step_names.append(field.name)
+        return step_names
 
 
 @dataclasses.dataclass(frozen=True)
-class CombinedStep(Generic[StartT, IntermediateT, EndT]):
+class StepSequence(ChainableWorkflowMixin[StartT, EndT]):
     """
     Composable workflow of single input callables.
 
@@ -106,33 +177,45 @@ class CombinedStep(Generic[StartT, IntermediateT, EndT]):
     >>> def stringify(x: float) -> str:
     ...    return str(x)
 
-    >>> CombinedStep(  # workflow (int -> float -> str)
-    ...    first=CombinedStep(  # workflow (int -> int -> float)
-    ...        first=plus_one,
-    ...        second=plus_half
-    ...    ),
-    ...    second=stringify
-    ... )(73)
-    '74.5'
-
-    >>> # is exactly equivalent to
-    >>> CombinedStep(first=plus_one, second=plus_half).chain(stringify)(73)
+    >>> StepSequence.start(plus_one).chain(plus_half).chain(stringify)(73)
     '74.5'
 
     """
 
-    first: Workflow[StartT, IntermediateT]
-    second: Workflow[IntermediateT, EndT]
+    @dataclasses.dataclass(frozen=True)
+    class __Steps:
+        inner: tuple[Workflow[Any, Any], ...]
+
+    # todo(ricoh): replace with normal tuple with TypeVarTuple hints
+    #   to enable automatic deduction StartT and EndT fom constructor
+    #   calls. TypeVarTuple is available in typing_extensions in
+    #   Python <= 3.11. Revise after mypy constraint is > 1.0.1,
+    #   which fails on trying to check TypeVarTuple.
+    steps: __Steps
 
     def __call__(self, inp: StartT) -> EndT:
-        return self.second(self.first(inp))
+        step_result: Any = inp
+        for step in self.steps.inner:
+            step_result = step(step_result)
+        return step_result
 
-    def chain(self, step: Workflow[EndT, NewEndT]) -> CombinedStep[StartT, EndT, NewEndT]:
-        return CombinedStep(first=self, second=step)
+    def chain(self, next_step: Workflow[EndT, NewEndT]) -> ChainableWorkflowMixin[StartT, NewEndT]:
+        return typing.cast(
+            ChainableWorkflowMixin[StartT, NewEndT],
+            self.__class__(self.__Steps((*self.steps.inner, next_step))),
+        )
+
+    @classmethod
+    def start(cls, first_step: Workflow[StartT, EndT]) -> ChainableWorkflowMixin[StartT, EndT]:
+        return cls(cls.__Steps((first_step,)))
 
 
 @dataclasses.dataclass(frozen=True)
-class CachedStep(Step[StartT, EndT], Generic[StartT, EndT, HashT]):
+class CachedStep(
+    ChainableWorkflowMixin[StartT, EndT],
+    ReplaceEnabledWorkflowMixin[StartT, EndT],
+    Generic[StartT, EndT, HashT],
+):
     """
     Cached workflow of single input callables.
 
@@ -157,11 +240,13 @@ class CachedStep(Step[StartT, EndT], Generic[StartT, EndT, HashT]):
     1
     """
 
+    step: Workflow[StartT, EndT]
     hash_function: Callable[[StartT], HashT] = dataclasses.field(default=hash)  # type: ignore[assignment]
 
     _cache: dict[HashT, EndT] = dataclasses.field(repr=False, init=False, default_factory=dict)
 
     def __call__(self, inp: StartT) -> EndT:
+        """Run the step only if the input is not cached, else return from cache."""
         hash_ = self.hash_function(inp)
         try:
             result = self._cache[hash_]
diff --git a/src/gt4py/next/program_processors/codegens/gtfn/gtfn_module.py b/src/gt4py/next/program_processors/codegens/gtfn/gtfn_module.py
index 18237c7f..91330243 100644
--- a/src/gt4py/next/program_processors/codegens/gtfn/gtfn_module.py
+++ b/src/gt4py/next/program_processors/codegens/gtfn/gtfn_module.py
@@ -12,6 +12,8 @@
 #
 # SPDX-License-Identifier: GPL-3.0-or-later
 
+from __future__ import annotations
+
 import dataclasses
 from typing import Any, Final, TypeVar
 
@@ -37,6 +39,10 @@ def get_param_description(name: str, obj: Any) -> interface.Parameter:
 
 @dataclasses.dataclass(frozen=True)
 class GTFNTranslationStep(
+    workflow.ChainableWorkflowMixin[
+        stages.ProgramCall,
+        stages.ProgramSource[languages.Cpp, languages.LanguageWithHeaderFilesSettings],
+    ],
     step_types.TranslationStep[languages.Cpp, languages.LanguageWithHeaderFilesSettings],
 ):
     language_settings: languages.LanguageWithHeaderFilesSettings = cpp_interface.CPP_DEFAULT
@@ -182,18 +188,6 @@ class GTFNTranslationStep(
         )
         return module
 
-    def chain(
-        self,
-        step: workflow.Workflow[
-            stages.ProgramSource[languages.Cpp, languages.LanguageWithHeaderFilesSettings], T
-        ],
-    ) -> workflow.CombinedStep[
-        stages.ProgramCall,
-        stages.ProgramSource[languages.Cpp, languages.LanguageWithHeaderFilesSettings],
-        T,
-    ]:
-        return workflow.CombinedStep(first=self, second=step)
-
 
 translate_program: Final[
     step_types.TranslationStep[languages.Cpp, languages.LanguageWithHeaderFilesSettings]
diff --git a/src/gt4py/next/program_processors/otf_compile_executor.py b/src/gt4py/next/program_processors/otf_compile_executor.py
new file mode 100644
index 00000000..a2202841
--- /dev/null
+++ b/src/gt4py/next/program_processors/otf_compile_executor.py
@@ -0,0 +1,56 @@
+# GT4Py - GridTools Framework
+#
+# Copyright (c) 2014-2023, ETH Zurich
+# All rights reserved.
+#
+# This file is part of the GT4Py project and the GridTools framework.
+# GT4Py is free software: you can redistribute it and/or modify it under
+# the terms of the GNU General Public License as published by the
+# Free Software Foundation, either version 3 of the License, or any later
+# version. See the LICENSE.txt file at the top-level directory of this
+# distribution for a copy of the license or check <https://www.gnu.org/licenses/>.
+#
+# SPDX-License-Identifier: GPL-3.0-or-later
+
+import dataclasses
+from typing import Any, Generic, Optional, TypeVar
+
+from gt4py.next.iterator import ir as itir
+from gt4py.next.otf import languages, recipes, stages, workflow
+from gt4py.next.program_processors import processor_interface as ppi
+
+
+SrcL = TypeVar("SrcL", bound=languages.LanguageTag)
+TgtL = TypeVar("TgtL", bound=languages.LanguageTag)
+LS = TypeVar("LS", bound=languages.LanguageSettings)
+HashT = TypeVar("HashT")
+
+
+@dataclasses.dataclass(frozen=True)
+class OTFCompileExecutor(ppi.ProgramExecutor, Generic[SrcL, LS, TgtL, HashT]):
+    otf_workflow: recipes.OTFCompileWorkflow[SrcL, LS, TgtL]
+    name: Optional[str] = None
+
+    def __call__(self, program: itir.FencilDefinition, *args, **kwargs: Any) -> None:
+        self.otf_workflow(stages.ProgramCall(program, args, kwargs))(
+            *args, offset_provider=kwargs["offset_provider"]
+        )
+
+    @property
+    def __name__(self) -> str:
+        return self.name or repr(self)
+
+
+@dataclasses.dataclass(frozen=True)
+class CachedOTFCompileExecutor(ppi.ProgramExecutor, Generic[SrcL, LS, TgtL, HashT]):
+    otf_workflow: workflow.CachedStep[stages.ProgramCall, stages.CompiledProgram, HashT]
+    name: Optional[str] = None
+
+    def __call__(self, program: itir.FencilDefinition, *args, **kwargs: Any) -> None:
+        self.otf_workflow(stages.ProgramCall(program, args, kwargs))(
+            *args, offset_provider=kwargs["offset_provider"]
+        )
+
+    @property
+    def __name__(self) -> str:
+        return self.name or repr(self)
diff --git a/src/gt4py/next/program_processors/runners/gtfn_cpu.py b/src/gt4py/next/program_processors/runners/gtfn_cpu.py
index 966fc784..3800380f 100644
--- a/src/gt4py/next/program_processors/runners/gtfn_cpu.py
+++ b/src/gt4py/next/program_processors/runners/gtfn_cpu.py
@@ -12,21 +12,18 @@
 #
 # SPDX-License-Identifier: GPL-3.0-or-later
 
-import dataclasses
-from typing import Any, Callable, Final, Optional
+from typing import Any
 
 import numpy as np
 import numpy.typing as npt
 
 from gt4py.eve.utils import content_hash
 from gt4py.next import common
-from gt4py.next.iterator import ir as itir
-from gt4py.next.otf import languages, stages, workflow
+from gt4py.next.otf import languages, recipes, stages, workflow
 from gt4py.next.otf.binding import cpp_interface, pybind
 from gt4py.next.otf.compilation import cache, compiler
 from gt4py.next.otf.compilation.build_systems import compiledb
-from gt4py.next.otf.workflow import CachedStep
-from gt4py.next.program_processors import processor_interface as ppi
+from gt4py.next.program_processors import otf_compile_executor
 from gt4py.next.program_processors.codegens.gtfn import gtfn_module
 from gt4py.next.type_system.type_translation import from_value
 
@@ -41,7 +38,7 @@ def convert_arg(arg: Any) -> Any:
         return arg
 
 
-def convert_args(inp: Callable) -> Callable:
+def convert_args(inp: stages.CompiledProgram) -> stages.CompiledProgram:
     def decorated_program(
         *args, offset_provider: dict[str, common.Connectivity | common.Dimension]
     ):
@@ -90,69 +87,39 @@ def compilation_hash(otf_closure: stages.ProgramCall) -> int:
     )
 
 
-@dataclasses.dataclass(frozen=True)
-class GTFNExecutor(ppi.ProgramExecutor):
-    language_settings: languages.LanguageWithHeaderFilesSettings = cpp_interface.CPP_DEFAULT
-    builder_factory: compiler.BuildSystemProjectGenerator = compiledb.CompiledbFactory()
-
-    name: Optional[str] = None
-    enable_itir_transforms: bool = True  # TODO replace by more general mechanism, see https://github.com/GridTools/gt4py/issues/1135
-    use_imperative_backend: bool = False
-
-    # TODO(tehrengruber): Revisit default value. The hash function for the caching currently
-    #  only uses a subset of the closure and implicitly relies on the workflow to only use that
-    #  information. As this is dangerous the caching is disabled by default and should be considered
-    #  experimental.
-    use_caching: bool = False
-    caching_strategy = cache.Strategy.SESSION
-
-    _otf_workflow: workflow.Workflow[
-        stages.ProgramCall, stages.CompiledProgram
-    ] = dataclasses.field(repr=False, init=False)
-
-    def __post_init__(self):
-        # TODO(tehrengruber): Restrict arguments of OTF workflow to the parts it actually needs
-        #  to compile the program instead of the entire closure.
-        otf_workflow: workflow.Workflow[stages.ProgramCall, stages.CompiledProgram] = (
-            gtfn_module.GTFNTranslationStep(
-                self.language_settings, self.enable_itir_transforms, self.use_imperative_backend
-            )
-            .chain(pybind.bind_source)
-            .chain(
-                compiler.Compiler(
-                    cache_strategy=self.caching_strategy, builder_factory=self.builder_factory
-                )
-            )
-            .chain(convert_args)
-        )
-
-        if self.use_caching:
-            otf_workflow = CachedStep(step=otf_workflow, hash_function=compilation_hash)
-
-        super().__setattr__("_otf_workflow", otf_workflow)
-
-    def __call__(self, program: itir.FencilDefinition, *args: Any, **kwargs: Any) -> None:
-        """
-        Execute the iterator IR program with the provided arguments.
+GTFN_DEFAULT_TRANSLATION_STEP = gtfn_module.GTFNTranslationStep(
+    cpp_interface.CPP_DEFAULT, enable_itir_transforms=True, use_imperative_backend=False
+)
 
-        The program is compiled to machine code with C++ as an intermediate step,
-        so the first execution is expected to have a significant overhead, while subsequent
-        calls are very fast. Only scalar and buffer arguments are supported currently.
+GTFN_DEFAULT_COMPILE_STEP = compiler.Compiler(
+    cache_strategy=cache.Strategy.SESSION, builder_factory=compiledb.CompiledbFactory()
+)
 
-        See ``ProgramExecutorFunction`` for details.
-        """
-        compiled_runner = self._otf_workflow(stages.ProgramCall(program, args, kwargs))
 
-        return compiled_runner(*args, offset_provider=kwargs["offset_provider"])
+GTFN_DEFAULT_WORKFLOW = recipes.OTFCompileWorkflow(
+    translation=GTFN_DEFAULT_TRANSLATION_STEP,
+    bindings=pybind.bind_source,
+    compilation=GTFN_DEFAULT_COMPILE_STEP,
+    decoration=convert_args,
+)
 
-    @property
-    def __name__(self) -> str:
-        return self.name or repr(self)
 
+run_gtfn = otf_compile_executor.OTFCompileExecutor[
+    languages.Cpp, languages.LanguageWithHeaderFilesSettings, languages.Python, Any
+](name="run_gtfn", otf_workflow=GTFN_DEFAULT_WORKFLOW)
 
-run_gtfn: Final[ppi.ProgramProcessor[None, ppi.ProgramExecutor]] = GTFNExecutor(
-    name="run_gtfn", use_imperative_backend=False
-)
-run_gtfn_imperative: Final[ppi.ProgramProcessor[None, ppi.ProgramExecutor]] = GTFNExecutor(
-    name="run_gtfn_imperative", use_imperative_backend=True
+run_gtfn_imperative = otf_compile_executor.OTFCompileExecutor[
+    languages.Cpp, languages.LanguageWithHeaderFilesSettings, languages.Python, Any
+](
+    name="run_gtfn_imperative",
+    otf_workflow=run_gtfn.otf_workflow.replace(
+        translation=run_gtfn.otf_workflow.translation.replace(use_imperative_backend=True),
+    ),
 )
+
+run_gtfn_cached = otf_compile_executor.CachedOTFCompileExecutor[
+    languages.Cpp, languages.LanguageWithHeaderFilesSettings, languages.Python, Any
+](
+    name="run_gtfn_cached",
+    otf_workflow=workflow.CachedStep(step=run_gtfn.otf_workflow, hash_function=compilation_hash),
+)  # todo(ricoh): add API for converting an executor to a cached version of itself and vice versa


Test patch:
diff --git a/tests/next_tests/integration_tests/feature_tests/iterator_tests/test_builtins.py b/tests/next_tests/integration_tests/feature_tests/iterator_tests/test_builtins.py
index 8bf4d867..d9c607b9 100644
--- a/tests/next_tests/integration_tests/feature_tests/iterator_tests/test_builtins.py
+++ b/tests/next_tests/integration_tests/feature_tests/iterator_tests/test_builtins.py
@@ -12,6 +12,7 @@
 #
 # SPDX-License-Identifier: GPL-3.0-or-later
 
+import dataclasses
 import math
 from typing import Callable, Iterable
 
@@ -51,7 +52,8 @@ from gt4py.next.iterator.builtins import (
 )
 from gt4py.next.iterator.embedded import NeighborTableOffsetProvider, np_as_located_field
 from gt4py.next.iterator.runtime import CartesianAxis, closure, fendef, fundef, offset
-from gt4py.next.program_processors.runners.gtfn_cpu import GTFNExecutor
+from gt4py.next.otf import workflow
+from gt4py.next.program_processors.runners.gtfn_cpu import run_gtfn
 
 from next_tests.integration_tests.feature_tests.math_builtin_test_data import math_builtin_test_data
 from next_tests.unit_tests.conftest import program_processor, run_processor
@@ -186,15 +188,13 @@ def test_arithmetic_and_logical_functors_gtfn(builtin, inputs, expected):
     inps = asfield(*asarray(*inputs))
     out = asfield((np.zeros_like(*asarray(expected))))[0]
 
-    gtfn_without_transforms = GTFNExecutor(
-        name="run_gtfn", enable_itir_transforms=False
+    gtfn_without_transforms = dataclasses.replace(
+        run_gtfn,
+        otf_workflow=run_gtfn.otf_workflow.replace(
+            translation=run_gtfn.otf_workflow.translation.replace(enable_itir_transforms=False),
+        ),
     )  # avoid inlining the function
-    fencil(
-        builtin,
-        out,
-        *inps,
-        processor=GTFNExecutor(name="run_gtfn", enable_itir_transforms=False),
-    )
+    fencil(builtin, out, *inps, processor=gtfn_without_transforms)
 
     assert np.allclose(np.asarray(out), expected)
 
diff --git a/tests/next_tests/integration_tests/feature_tests/otf_tests/test_pybind_build.py b/tests/next_tests/integration_tests/feature_tests/otf_tests/test_pybind_build.py
index aac1947c..4eee84c8 100644
--- a/tests/next_tests/integration_tests/feature_tests/otf_tests/test_pybind_build.py
+++ b/tests/next_tests/integration_tests/feature_tests/otf_tests/test_pybind_build.py
@@ -28,7 +28,7 @@ from next_tests.unit_tests.otf_tests.compilation_tests.build_systems_tests.conft
 
 def test_gtfn_cpp_with_cmake(program_source_with_name):
     example_program_source = program_source_with_name("gtfn_cpp_with_cmake")
-    build_the_program = workflow.Step(pybind.bind_source).chain(
+    build_the_program = workflow.make_step(pybind.bind_source).chain(
         compiler.Compiler(
             cache_strategy=cache.Strategy.SESSION, builder_factory=cmake.CMakeFactory()
         ),
@@ -43,7 +43,7 @@ def test_gtfn_cpp_with_cmake(program_source_with_name):
 
 def test_gtfn_cpp_with_compiledb(program_source_with_name):
     example_program_source = program_source_with_name("gtfn_cpp_with_compiledb")
-    build_the_program = workflow.Step(pybind.bind_source).chain(
+    build_the_program = workflow.make_step(pybind.bind_source).chain(
         compiler.Compiler(
             cache_strategy=cache.Strategy.SESSION,
             builder_factory=compiledb.CompiledbFactory(),
diff --git a/tests/next_tests/unit_tests/otf_tests/test_workflow.py b/tests/next_tests/unit_tests/otf_tests/test_workflow.py
index c6b18b77..9e1e14ed 100644
--- a/tests/next_tests/unit_tests/otf_tests/test_workflow.py
+++ b/tests/next_tests/unit_tests/otf_tests/test_workflow.py
@@ -28,8 +28,15 @@ class StageTwo:
     y: str
 
 
-def step_zero(inp: int) -> StageOne:
-    return StageOne(inp)
+@dataclasses.dataclass(frozen=True)
+class NamedStepsExample(workflow.NamedStepSequence[int, str]):
+    repeat: workflow.Workflow[int, list[int]]
+    strify: workflow.Workflow[list[int], str]
+
+
+@dataclasses.dataclass(frozen=True)
+class SingleStep(workflow.NamedStepSequence[int, StageTwo]):
+    step: workflow.Workflow[int, StageTwo]
 
 
 def step_one(inp: StageOne) -> StageTwo:
@@ -41,17 +48,43 @@ def step_two(inp: StageTwo) -> str:
 
 
 def test_single_step():
-    step1: workflow.Workflow[StageOne, StageTwo] = workflow.Step(step_one)
+    step1: workflow.Workflow[StageOne, StageTwo] = workflow.make_step(step_one)
     assert step1(StageOne(3)) == step_one(StageOne(3))
 
 
-def test_chain_step_step():
-    wf: workflow.Workflow[StageOne, str] = workflow.Step(step_one).chain(step_two)
+def test_chain_step_sequence():
+    wf: workflow.Workflow[StageOne, str] = workflow.StepSequence.start(step_one).chain(step_two)
     inp = StageOne(5)
     assert wf(inp) == step_two(step_one(inp))
 
 
-def test_chain_combinedstep_step():
-    initial_workflow: workflow.Workflow[int, StageTwo] = workflow.CombinedStep(step_zero, step_one)
-    full_workflow: workflow.Workflow[int, str] = initial_workflow.chain(step_two)
-    assert full_workflow(42) == "42"
+def test_named_steps():
+    """Test composing named steps"""
+
+    wf = NamedStepsExample(repeat=lambda inp: [inp] * 3, strify=lambda inp: str(inp))
+    assert wf.repeat(4) == [4, 4, 4]
+    assert wf.strify([1, 2, 3]) == "[1, 2, 3]"
+    assert wf(4) == "[4, 4, 4]"
+
+
+def test_chain_from_named():
+    initial_workflow: workflow.Workflow[StageOne, StageTwo] = SingleStep(step=step_one)
+    full_workflow: workflow.Workflow[StageOne, str] = initial_workflow.chain(step_two)
+    assert full_workflow(StageOne(42)) == "42"
+
+
+def test_cached_with_hashing():
+    def hashing(inp: list[int]) -> int:
+        return hash(sum(inp))
+
+    wf = workflow.CachedStep(step=lambda inp: inp + [1], hash_function=hashing)
+
+    assert wf([1, 2, 3]) == [1, 2, 3, 1]
+    assert wf([3, 2, 1]) == [1, 2, 3, 1]
+
+
+def test_replace():
+    """Test replacing a named step."""
+    wf = NamedStepsExample(repeat=lambda inp: [inp] * 3, strify=lambda inp: str(inp))
+    wf_repl = wf.replace(repeat=lambda inp: [inp] * 4)
+    assert wf_repl(4) == "[4, 4, 4, 4]"