Repository: GridTools/gt4py
Pull request #1227 (2023-04-04 02:14:26)

feature[eve]: Add functionality to preserve parts of the annex in `NodeTranslator`s
Allows preserving parts of the annex using the `PRESERVED_ANNEX_ATTRS` class variable of a `NodeTranslator`.

@egparedes I was a little surprised to not find any tests for `NodeVisitor`s or `NodeTranslator`s. Am I missing something? Otherwise we should add some in the future.


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/eve/concepts.py b/src/gt4py/eve/concepts.py
index 36361d79..d66e8583 100644
--- a/src/gt4py/eve/concepts.py
+++ b/src/gt4py/eve/concepts.py
@@ -197,9 +197,12 @@ class Node(datamodels.DataModel, trees.Tree, kw_only=True):  # type: ignore[call
     comparisons of the node, since it is not really a field. Thus, visitors
     and pipeline passes can freely attach computed attributes into the instance
     `annex`. Note that `annex` attribute is not implicitly copied in the
-    :class:`NodeTranslator`. If you want it to persist accross multiple
-    :class:`NodeTranslation` use a `root_validator` to dynamically (re)compute
-    the annex on node construction (see e.g. :class:`SymbolTableTrait`).
+    :class:`NodeTranslator`. If you want it to persist across multiple
+    :class:`NodeTranslator`s either use a `root_validator` to dynamically
+    (re)compute the annex on node construction (see e.g.
+    :class:`SymbolTableTrait`) or add the parts of the annex that should be
+    preserved to the `PRESERVED_ANNEX_ATTRS` attribute in the
+    :class:`NodeTranslator` class..
     """
 
     __slots__ = ()
diff --git a/src/gt4py/eve/visitors.py b/src/gt4py/eve/visitors.py
index 34660fa2..fe5f9e14 100644
--- a/src/gt4py/eve/visitors.py
+++ b/src/gt4py/eve/visitors.py
@@ -19,6 +19,7 @@ from __future__ import annotations
 
 import collections.abc
 import copy
+from typing import ClassVar
 
 from . import concepts, trees
 from .extended_typing import Any
@@ -151,6 +152,8 @@ class NodeTranslator(NodeVisitor):
 
     """
 
+    PRESERVED_ANNEX_ATTRS: ClassVar[tuple[str, ...]] = ()
+
     def generic_visit(self, node: concepts.RootNode, **kwargs: Any) -> Any:
         memo = kwargs.get("__memo__", None)
 
@@ -162,6 +165,14 @@ class NodeTranslator(NodeVisitor):
                     if (new_child := self.visit(child, **kwargs)) is not NOTHING
                 },
             )
+            if self.PRESERVED_ANNEX_ATTRS and (old_annex := getattr(node, "__node_annex__", None)):
+                # access to `new_node.annex` implicitly creates the `__node_annex__` attribute in the property getter
+                new_annex_dict = new_node.annex.__dict__
+                for key in self.PRESERVED_ANNEX_ATTRS:
+                    if value := getattr(old_annex, key, None):
+                        assert key not in new_annex_dict
+                        new_annex_dict[key] = value
+
             return new_node
 
         if isinstance(node, (list, tuple, set, collections.abc.Set)) or (
diff --git a/src/gt4py/next/iterator/transforms/trace_shifts.py b/src/gt4py/next/iterator/transforms/trace_shifts.py
index 284fc2cc..5024c607 100644
--- a/src/gt4py/next/iterator/transforms/trace_shifts.py
+++ b/src/gt4py/next/iterator/transforms/trace_shifts.py
@@ -12,15 +12,20 @@
 #
 # SPDX-License-Identifier: GPL-3.0-or-later
 
+import types
 from collections.abc import Callable
 from dataclasses import dataclass
-from typing import Any, Final
+from typing import Any, Final, Union
 
 from gt4py.eve import NodeTranslator
 from gt4py.next.iterator import ir
 from gt4py.next.iterator.transforms.collect_shifts import ALL_NEIGHBORS
 
 
+VALUE_TOKEN = types.new_class("VALUE_TOKEN")
+TYPE_TOKEN = types.new_class("TYPE_TOKEN")
+
+
 @dataclass(frozen=True)
 class InputTracer:
     inp: str
@@ -37,77 +42,26 @@ class InputTracer:
         )
 
     def deref(self):
-        if self.lift_level:
-            return self
         self.register_deref(self.inp, self.offsets)
+        return VALUE_TOKEN
 
-    def lift(self):
-        return InputTracer(
-            inp=self.inp,
-            register_deref=self.register_deref,
-            offsets=self.offsets,
-            lift_level=self.lift_level + 1,
-        )
-
-    def unlift(self):
-        assert self.lift_level > 0
-        return InputTracer(
-            inp=self.inp,
-            register_deref=self.register_deref,
-            offsets=self.offsets,
-            lift_level=self.lift_level - 1,
-        )
-
-
-@dataclass(frozen=True)
-class CombinedTracer:
-    tracers: tuple[InputTracer, ...]
-
-    def shift(self, offsets):
-        return CombinedTracer(tracers=tuple(t.shift(offsets) for t in self.tracers))
-
-    def deref(self):
-        lift_levels = {t.lift_level for t in self.tracers}
-        assert len(lift_levels) == 1
-        if lift_levels.pop():
-            return self
-        for t in self.tracers:
-            t.deref()
-
-    def lift(self):
-        return CombinedTracer(tracers=tuple(t.lift() for t in self.tracers))
-
-    def unlift(self):
-        return CombinedTracer(tracers=tuple(t.unlift() for t in self.tracers))
-
-
-def _combine(*tracers):
-    input_tracers = []
-
-    def handle_tracer(tracer):
-        if isinstance(tracer, InputTracer):
-            input_tracers.append(tracer)
-        elif isinstance(tracer, CombinedTracer):
-            input_tracers.extend(tracer.tracers)
-        elif isinstance(tracer, tuple):
-            for t in tracer:
-                handle_tracer(t)
 
-    for tracer in tracers:
-        handle_tracer(tracer)
-
-    return CombinedTracer(tracers=tuple(input_tracers))
+def _combine(*values):
+    # `OffsetLiteral`s may occur in `list_get` calls
+    if not all(
+        val in [VALUE_TOKEN, TYPE_TOKEN] or isinstance(val, ir.OffsetLiteral) for val in values
+    ):
+        raise AssertionError("All arguments must be values or types.")
+    return VALUE_TOKEN
 
 
 # implementations of builtins
-
-
 def _deref(x):
     return x.deref()
 
 
 def _can_deref(x):
-    return
+    return VALUE_TOKEN
 
 
 def _shift(*offsets):
@@ -117,9 +71,23 @@ def _shift(*offsets):
     return apply
 
 
+@dataclass(frozen=True)
+class AppliedLift:
+    stencil: Callable
+    its: tuple[Union[InputTracer, "AppliedLift"]]
+
+    def shift(self, offsets):
+        return AppliedLift(self.stencil, tuple(_shift(it) for it in self.its))
+
+    def deref(self):
+        return self.stencil(*self.its)
+
+
 def _lift(f):
-    def apply(*args):
-        return f(*(arg.lift() for arg in args)).unlift()
+    def apply(*its):
+        if not all(isinstance(it, (InputTracer, AppliedLift)) for it in its):
+            raise AssertionError("All arguments must be iterators.")
+        return AppliedLift(f, its)
 
     return apply
 
@@ -128,8 +96,12 @@ def _reduce(f, init):
     return _combine
 
 
+def _map(f):
+    return _combine
+
+
 def _neighbors(o, x):
-    return x.shift((o, ALL_NEIGHBORS)).deref()
+    return _deref(_shift(o, ALL_NEIGHBORS)(x))
 
 
 def _scan(f, forward, init):
@@ -139,32 +111,27 @@ def _scan(f, forward, init):
     return apply
 
 
-def _make_tuple(*args):
-    return args
-
-
-def _tuple_get(idx, tup):
-    if tup is not None:
-        return tup[int(idx.value)]
-
-
 _START_CTX: Final = {
     "deref": _deref,
     "can_deref": _can_deref,
     "shift": _shift,
     "lift": _lift,
-    "reduce": _reduce,
     "scan": _scan,
-    "make_tuple": _make_tuple,
-    "tuple_get": _tuple_get,
+    "reduce": _reduce,
     "neighbors": _neighbors,
+    "map_": _map,
 }
 
 
 class TraceShifts(NodeTranslator):
+    def visit_Literal(self, node: ir.SymRef, *, ctx: dict[str, Any]) -> Any:
+        return VALUE_TOKEN
+
     def visit_SymRef(self, node: ir.SymRef, *, ctx: dict[str, Any]) -> Any:
         if node.id in ctx:
             return ctx[node.id]
+        elif node.id in ir.TYPEBUILTINS:
+            return TYPE_TOKEN
         return _combine
 
     def visit_FunCall(self, node: ir.FunCall, *, ctx: dict[str, Any]) -> Any:
@@ -182,10 +149,15 @@ class TraceShifts(NodeTranslator):
         self, node: ir.StencilClosure, *, shifts: dict[str, list[tuple[ir.OffsetLiteral, ...]]]
     ):
         def register_deref(inp: str, offsets: tuple[ir.OffsetLiteral, ...]):
-            shifts.setdefault(inp, []).append(offsets)
+            shifts[inp].append(offsets)
+
+        tracers = []
+        for inp in node.inputs:
+            shifts.setdefault(inp.id, [])
+            tracers.append(InputTracer(inp=inp.id, register_deref=register_deref))
 
-        tracers = [InputTracer(inp=inp.id, register_deref=register_deref) for inp in node.inputs]
-        self.visit(node.stencil, ctx=_START_CTX)(*tracers)
+        result = self.visit(node.stencil, ctx=_START_CTX)(*tracers)
+        assert result is VALUE_TOKEN
 
     @classmethod
     def apply(cls, node: ir.StencilClosure) -> dict[str, list[tuple[ir.OffsetLiteral, ...]]]:
diff --git a/src/gt4py/next/iterator/type_inference.py b/src/gt4py/next/iterator/type_inference.py
index a1594fb7..ae783217 100644
--- a/src/gt4py/next/iterator/type_inference.py
+++ b/src/gt4py/next/iterator/type_inference.py
@@ -718,7 +718,12 @@ class _TypeInferrer(eve.traits.VisitorWithSymbolTableTrait, eve.NodeTranslator):
 
     def _visit_shift(self, node: ir.FunCall, **kwargs) -> Type:
         # Calls to shift are handled as being part of the grammar, not
-        # as function calls, as the type depends on the offset provider
+        # as function calls, as the type depends on the offset provider.
+
+        # Visit arguments such that their type is also inferred (particularly important for
+        # dynamic offsets)
+        self.visit(node.args)
+
         current_loc_in, current_loc_out = _infer_shift_location_types(
             node.args, self.offset_provider, self.constraints
         )
diff --git a/src/gt4py/next/otf/binding/cpp_interface.py b/src/gt4py/next/otf/binding/cpp_interface.py
index 5b8a1931..1c18f812 100644
--- a/src/gt4py/next/otf/binding/cpp_interface.py
+++ b/src/gt4py/next/otf/binding/cpp_interface.py
@@ -14,9 +14,9 @@
 
 from typing import Final, Sequence
 
-import gt4py.next.type_system.type_specifications as ts
 from gt4py.next.otf import languages
 from gt4py.next.otf.binding import interface
+from gt4py.next.type_system import type_info as ti, type_specifications as ts
 
 
 CPP_DEFAULT: Final = languages.LanguageWithHeaderFilesSettings(
@@ -52,7 +52,9 @@ def render_scalar_type(scalar_type: ts.ScalarType) -> str:
 def _render_function_param(param: interface.Parameter, index: int) -> str:
     if isinstance(param.type_, ts.ScalarType):
         return f"{render_scalar_type(param.type_)} {param.name}"
-    elif isinstance(param.type_, ts.FieldType):
+    elif ti.is_type_or_tuple_of_type(
+        param.type_, ts.FieldType
+    ):  # TODO(havogt): add support for scalar tuples
         return f"BufferT{index}&& {param.name}"
     else:
         raise ValueError(f"Type '{param.type_}' is not supported in C++ interfaces.")
@@ -68,7 +70,9 @@ def render_function_declaration(function: interface.Function, body: str) -> str:
     template_params = [
         f"class BufferT{index}"
         for index, param in enumerate(function.parameters)
-        if isinstance(param.type_, ts.FieldType)
+        if ti.is_type_or_tuple_of_type(
+            param.type_, ts.FieldType
+        )  # TODO(havogt): add support for scalar tuples
     ]
     if template_params:
         return f"""
diff --git a/src/gt4py/next/otf/binding/pybind.py b/src/gt4py/next/otf/binding/pybind.py
index f691b31f..5ba229a8 100644
--- a/src/gt4py/next/otf/binding/pybind.py
+++ b/src/gt4py/next/otf/binding/pybind.py
@@ -17,13 +17,13 @@
 
 from __future__ import annotations
 
-from typing import Any, Sequence
+from typing import Any, Sequence, Union
 
 import gt4py.eve as eve
-import gt4py.next.type_system.type_specifications as ts
 from gt4py.eve.codegen import JinjaTemplate as as_jinja, TemplatedGenerator
 from gt4py.next.otf import languages, stages, workflow
 from gt4py.next.otf.binding import cpp_interface, interface
+from gt4py.next.type_system import type_info as ti, type_specifications as ts
 
 
 class Expr(eve.Node):
@@ -42,6 +42,10 @@ class BufferSID(Expr):
     # unit_stride_dim: int # TODO(havogt): we can fix the dimension with unity stride once we have the "frozen stencil" mechanism
 
 
+class CompositeSID(Expr):
+    elems: Sequence[Union[BufferSID, CompositeSID]]
+
+
 class FunctionCall(Expr):
     target: interface.Function
     args: Sequence[Any]
@@ -81,6 +85,17 @@ class BindingFile(eve.Node):
     binding_module: BindingModule
 
 
+def _type_string(type_: ts.TypeSpec) -> str:
+    if isinstance(type_, ts.TupleType):
+        return f"std::tuple<{','.join(_type_string(t) for t in type_.types)}>"
+    elif isinstance(type_, ts.FieldType):
+        return "pybind11::buffer"
+    elif isinstance(type_, ts.ScalarType):
+        return cpp_interface.render_scalar_type(type_)
+    else:
+        raise ValueError(f"Type '{type_}' is not supported in pybind11 interfaces.")
+
+
 class BindingCodeGenerator(TemplatedGenerator):
     BindingFile = as_jinja(
         """\
@@ -107,14 +122,7 @@ class BindingCodeGenerator(TemplatedGenerator):
         """
     )
 
-    def visit_FunctionParameter(self, param: FunctionParameter):
-        if isinstance(param.type_, ts.FieldType):
-            type_str = "pybind11::buffer"
-        elif isinstance(param.type_, ts.ScalarType):
-            type_str = cpp_interface.render_scalar_type(param.type_)
-        else:
-            raise ValueError(f"Type '{param.type_}' is not supported in pybind11 interfaces.")
-        return f"{type_str} {param.name}"
+    FunctionParameter = as_jinja("{{_this_module._type_string(_this_node.type_)}} {{name}}")
 
     ReturnStmt = as_jinja("""return {{expr}};""")
 
@@ -133,7 +141,7 @@ class BindingCodeGenerator(TemplatedGenerator):
         args = [self.visit(arg) for arg in call.args]
         return cpp_interface.render_function_call(call.target, args)
 
-    def visit_BufferSID(self, sid: BufferSID):
+    def visit_BufferSID(self, sid: BufferSID, **kwargs):
         return self.generic_visit(
             sid, rendered_scalar_type=cpp_interface.render_scalar_type(sid.scalar_type)
         )
@@ -146,18 +154,38 @@ class BindingCodeGenerator(TemplatedGenerator):
             )"""
     )
 
+    def visit_CompositeSID(self, node: CompositeSID, **kwargs):
+        kwargs["composite_ids"] = (
+            f"gridtools::integral_constant<int,{i}>" for i in range(len(node.elems))
+        )
+        return self.generic_visit(node, **kwargs)
+
+    CompositeSID = as_jinja(
+        "gridtools::sid::composite::keys<{{','.join(composite_ids)}}>::make_values({{','.join(elems)}})"
+    )
+
     DimensionType = as_jinja("""generated::{{name}}_t""")
 
 
-def make_argument(param: interface.Parameter) -> str | BufferSID:
-    if isinstance(param.type_, ts.FieldType):
+def _tuple_get(index: int, var: str) -> str:
+    return f"gridtools::tuple_util::get<{index}>({var})"
+
+
+def make_argument(name: str, type_: ts.TypeSpec) -> str | BufferSID | CompositeSID:
+    if isinstance(type_, ts.FieldType):
         return BufferSID(
-            source_buffer=param.name,
-            dimensions=[DimensionType(name=dim.value) for dim in param.type_.dims],
-            scalar_type=param.type_.dtype,
+            source_buffer=name,
+            dimensions=[DimensionType(name=dim.value) for dim in type_.dims],
+            scalar_type=type_.dtype,
+        )
+    elif ti.is_tuple_of_type(type_, ts.FieldType):
+        return CompositeSID(
+            elems=[make_argument(_tuple_get(i, name), t) for i, t in enumerate(type_.types)]
         )
+    elif isinstance(type_, ts.ScalarType):
+        return name
     else:
-        return param.name
+        raise ValueError(f"Type '{type_}' is not supported in pybind11 interfaces.")
 
 
 def create_bindings(
@@ -183,9 +211,11 @@ def create_bindings(
             "pybind11/pybind11.h",
             "pybind11/stl.h",
             "gridtools/storage/adapter/python_sid_adapter.hpp",
+            "gridtools/sid/composite.hpp",
             "gridtools/sid/unknown_kind.hpp",
             "gridtools/sid/rename_dimensions.hpp",
             "gridtools/common/defs.hpp",
+            "gridtools/common/tuple_util.hpp",
             "gridtools/fn/unstructured.hpp",
             "gridtools/fn/cartesian.hpp",
             "gridtools/fn/backend/naive.hpp",
@@ -199,7 +229,10 @@ def create_bindings(
             body=ReturnStmt(
                 expr=FunctionCall(
                     target=program_source.entry_point,
-                    args=[make_argument(param) for param in program_source.entry_point.parameters],
+                    args=[
+                        make_argument(param.name, param.type_)
+                        for param in program_source.entry_point.parameters
+                    ],
                 )
             ),
         ),
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 d257b6a2..3800380f 100644
--- a/src/gt4py/next/program_processors/runners/gtfn_cpu.py
+++ b/src/gt4py/next/program_processors/runners/gtfn_cpu.py
@@ -12,34 +12,33 @@
 #
 # 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
 
 
 # TODO(ricoh): Add support for the whole range of arguments that can be passed to a fencil.
 def convert_arg(arg: Any) -> Any:
+    if isinstance(arg, tuple):
+        return tuple(convert_arg(a) for a in arg)
     if hasattr(arg, "__array__"):
         return np.asarray(arg)
     else:
         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]
     ):
@@ -88,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
diff --git a/src/gt4py/next/type_system/type_info.py b/src/gt4py/next/type_system/type_info.py
index 1936841e..6bb3622b 100644
--- a/src/gt4py/next/type_system/type_info.py
+++ b/src/gt4py/next/type_system/type_info.py
@@ -241,7 +241,7 @@ def arithmetic_bounds(arithmetic_type: ts.ScalarType):
 
 def is_type_or_tuple_of_type(type_: ts.TypeSpec, expected_type: type | tuple) -> bool:
     """
-    Return True if ``type_`` matches any of the expected.
+    Return True if ``type_`` matches any of the expected or is a tuple of them.
 
     Examples:
     ---------
@@ -259,6 +259,26 @@ def is_type_or_tuple_of_type(type_: ts.TypeSpec, expected_type: type | tuple) ->
     return all(isinstance(t, expected_type) for t in primitive_constituents(type_))
 
 
+def is_tuple_of_type(type_: ts.TypeSpec, expected_type: type | tuple) -> TypeGuard[ts.TupleType]:
+    """
+    Return True if ``type_`` matches (nested) tuple of ``expected_type``.
+
+    Examples:
+    ---------
+    >>> scalar_type = ts.ScalarType(kind=ts.ScalarKind.INT)
+    >>> field_type = ts.FieldType(dims=[], dtype=scalar_type)
+    >>> is_tuple_of_type(field_type, ts.FieldType)
+    False
+    >>> is_tuple_of_type(ts.TupleType(types=[scalar_type, field_type]), (ts.ScalarType, ts.FieldType))
+    True
+    >>> is_tuple_of_type(ts.TupleType(types=[scalar_type]), ts.FieldType)
+    False
+    >>> is_tuple_of_type(ts.TupleType(types=[scalar_type, field_type]), ts.FieldType)
+    False
+    """
+    return isinstance(type_, ts.TupleType) and is_type_or_tuple_of_type(type_, expected_type)
+
+
 def extract_dims(symbol_type: ts.TypeSpec) -> list[Dimension]:
     """
     Try to extract field dimensions if possible.


Test patch:
diff --git a/tests/eve_tests/unit_tests/test_visitors.py b/tests/eve_tests/unit_tests/test_visitors.py
new file mode 100644
index 00000000..70b7e2e4
--- /dev/null
+++ b/tests/eve_tests/unit_tests/test_visitors.py
@@ -0,0 +1,30 @@
+# 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
+
+from gt4py import eve
+
+
+def test_annex_preservation(compound_node: eve.Node):
+    compound_node.annex.foo = 1
+    compound_node.annex.bar = 2
+
+    class SampleTranslator(eve.NodeTranslator):
+        PRESERVED_ANNEX_ATTRS = ("foo",)
+
+    translated_node = SampleTranslator().visit(compound_node)
+
+    assert translated_node.annex.foo == 1
+    assert not hasattr(translated_node.annex, "bar")
diff --git a/tests/next_tests/integration_tests/feature_tests/ffront_tests/test_execution.py b/tests/next_tests/integration_tests/feature_tests/ffront_tests/test_execution.py
index 82fe7d5f..caac4093 100644
--- a/tests/next_tests/integration_tests/feature_tests/ffront_tests/test_execution.py
+++ b/tests/next_tests/integration_tests/feature_tests/ffront_tests/test_execution.py
@@ -26,8 +26,8 @@ from gt4py.next.ffront.fbuiltins import (
     Field,
     astype,
     broadcast,
+    float32,
     float64,
-    int32,
     int64,
     maximum,
     minimum,
@@ -57,26 +57,27 @@ def test_copy(fieldview_backend):
     assert np.allclose(a_I_float, b_I_float)
 
 
-@pytest.mark.skip(reason="no lowering for returning a tuple of fields exists yet.")
 def test_multicopy(fieldview_backend):
-    a_I_float = np_as_located_field(IDim)(np.random.randn(size).astype("float64"))
-    b_I_float = np_as_located_field(IDim)(np.random.randn(size).astype("float64"))
-    out_I_float = np_as_located_field(IDim)(np.zeros((size), dtype=float64))
-    out_I_float_1 = np_as_located_field(IDim)(np.zeros((size), dtype=float64))
+    inp0 = np_as_located_field(IDim)(np.random.randn(size).astype("float64"))
+    inp1 = np_as_located_field(IDim)(np.random.randn(size).astype("float32"))
+    out0 = np_as_located_field(IDim)(np.zeros((size), dtype=float64))
+    out1 = np_as_located_field(IDim)(np.zeros((size), dtype=float32))
 
     @field_operator(backend=fieldview_backend)
     def multicopy(
-        inp1: Field[[IDim], float64], inp2: Field[[IDim], float64]
-    ) -> tuple[Field[[IDim], float64], Field[[IDim], float64]]:
+        inp1: Field[[IDim], float64], inp2: Field[[IDim], float32]
+    ) -> tuple[Field[[IDim], float64], Field[[IDim], float32]]:
         return inp1, inp2
 
-    assert np.allclose(a_I_float, out_I_float)
-    assert np.allclose(b_I_float, out_I_float_1)
+    multicopy(inp0, inp1, out=(out0, out1), offset_provider={})
+
+    assert np.allclose(inp0, out0)
+    assert np.allclose(inp1, out1)
 
 
 def test_cartesian_shift(fieldview_backend):
     a = np_as_located_field(IDim)(np.arange(size + 1, dtype=np.float64))
-    out_I_float = np_as_located_field(IDim)(np.zeros((size), dtype=float64))
+    out_I_float = np_as_located_field(IDim)(np.zeros((size,), dtype=float64))
 
     @field_operator
     def shift_by_one(inp: Field[[IDim], float64]) -> Field[[IDim], float64]:
@@ -114,7 +115,7 @@ def test_fold_shifts(fieldview_backend):
     """Shifting the result of an addition should work."""
     a = np_as_located_field(IDim)(np.arange(size + 1, dtype=np.float64))
     b = np_as_located_field(IDim)(np.ones((size + 2)) * 2)
-    out_I_float = np_as_located_field(IDim)(np.zeros((size), dtype=float64))
+    out_I_float = np_as_located_field(IDim)(np.zeros((size,), dtype=float64))
 
     @field_operator
     def auto_lift(
@@ -137,7 +138,7 @@ def test_fold_shifts(fieldview_backend):
 def test_tuples(fieldview_backend):
     a_I_float = np_as_located_field(IDim)(np.random.randn(size).astype("float64"))
     b_I_float = np_as_located_field(IDim)(np.random.randn(size).astype("float64"))
-    out_I_float = np_as_located_field(IDim)(np.zeros((size), dtype=float64))
+    out_I_float = np_as_located_field(IDim)(np.zeros((size,), dtype=float64))
 
     @field_operator
     def tuples(
@@ -197,7 +198,7 @@ def test_scalar_arg_with_field(fieldview_backend):
 
     inp = index_field(Edge, dtype=float64)
     factor = 3.0
-    out = np_as_located_field(Edge)(np.zeros((size), dtype=np.float64))
+    out = np_as_located_field(Edge)(np.zeros((size,), dtype=np.float64))
 
     @field_operator
     def scalar_and_field_args(
@@ -261,7 +262,7 @@ def test_scalar_scan(fieldview_backend):
 
 def test_tuple_scalar_scan(fieldview_backend):
     if fieldview_backend in [gtfn_cpu.run_gtfn, gtfn_cpu.run_gtfn_imperative]:
-        pytest.skip("Tuple arguments are not supported in gtfn yet.")
+        pytest.skip("Scalar tuple arguments are not supported in gtfn yet.")
 
     size = 10
     KDim = Dimension("K", kind=DimensionKind.VERTICAL)
@@ -287,7 +288,7 @@ def test_tuple_scalar_scan(fieldview_backend):
 
 def test_astype_int(fieldview_backend):
     size = 10
-    b_float_64 = np_as_located_field(IDim)(np.ones((size), dtype=np.float64))
+    b_float_64 = np_as_located_field(IDim)(np.full((size,), fill_value=1.5, dtype=np.float64))
     c_int64 = np_as_located_field(IDim)(np.ones((size,), dtype=np.int64))
     out_int_64 = np_as_located_field(IDim)(np.zeros((size,), dtype=np.int64))
 
@@ -301,7 +302,7 @@ def test_astype_int(fieldview_backend):
 
 
 def test_astype_bool(fieldview_backend):
-    b_float_64 = np_as_located_field(IDim)(np.ones((size), dtype=np.float64))
+    b_float_64 = np_as_located_field(IDim)(np.full((size,), fill_value=0.5, dtype=np.float64))
     c_bool = np_as_located_field(IDim)(np.ones((size,), dtype=bool))
     out_bool = np_as_located_field(IDim)(np.zeros((size,), dtype=bool))
 
@@ -315,17 +316,18 @@ def test_astype_bool(fieldview_backend):
 
 
 def test_astype_float(fieldview_backend):
-    c_int64 = np_as_located_field(IDim)(np.ones((size,), dtype=np.int64))
-    c_int32 = np_as_located_field(IDim)(np.ones((size,), dtype=np.int32))
-    out_int_32 = np_as_located_field(IDim)(np.zeros((size,), dtype=np.int32))
+    c_float64 = np_as_located_field(IDim)(
+        np.full((size,), fill_value=np.float64("5e300"), dtype=np.float64)
+    )
+    out_int_32 = np_as_located_field(IDim)(np.zeros((size,), dtype=np.float32))
 
     @field_operator(backend=fieldview_backend)
-    def astype_fieldop_float(b: Field[[IDim], int64]) -> Field[[IDim], int32]:
-        d = astype(b, int32)
+    def astype_fieldop_float(b: Field[[IDim], float64]) -> Field[[IDim], np.float32]:
+        d = astype(b, float32)
         return d
 
-    astype_fieldop_float(c_int64, out=out_int_32, offset_provider={})
-    assert np.allclose(c_int32.array(), out_int_32)
+    astype_fieldop_float(c_float64, out=out_int_32, offset_provider={})
+    assert np.all(out_int_32.array() == np.inf)
 
 
 def test_offset_field(fieldview_backend):
@@ -334,7 +336,7 @@ def test_offset_field(fieldview_backend):
     a_I_float_1 = np_as_located_field(IDim, KDim)(
         np.append(np.insert(a_I_arr, size, 0, axis=1), [np.array([0] * (size + 1))], axis=0)
     )
-    offset_field_arr = np.asarray(np.ones((size - 1, size - 1)), dtype=int64)
+    offset_field_arr = np.ones((size - 1, size - 1), dtype=int64)
     offset_field_comp = np.append(
         np.insert(offset_field_arr, size - 1, 0, axis=1), [np.array([0] * size)], axis=0
     )
@@ -375,7 +377,7 @@ def test_offset_field(fieldview_backend):
 def test_nested_tuple_return(fieldview_backend):
     a_I_float = np_as_located_field(IDim)(np.random.randn(size).astype("float64"))
     b_I_float = np_as_located_field(IDim)(np.random.randn(size).astype("float64"))
-    out_I_float = np_as_located_field(IDim)(np.zeros((size), dtype=float64))
+    out_I_float = np_as_located_field(IDim)(np.zeros((size,), dtype=float64))
 
     @field_operator
     def pack_tuple(
@@ -507,10 +509,7 @@ def test_tuple_with_local_field_in_reduction_shifted(reduction_setup, fieldview_
 def test_tuple_arg(fieldview_backend):
     a_I_float = np_as_located_field(IDim)(np.random.randn(size).astype("float64"))
     b_I_float = np_as_located_field(IDim)(np.random.randn(size).astype("float64"))
-    out_I_float = np_as_located_field(IDim)(np.zeros((size), dtype=float64))
-
-    if fieldview_backend in [gtfn_cpu.run_gtfn, gtfn_cpu.run_gtfn_imperative]:
-        pytest.skip("Tuple arguments are not supported in gtfn yet.")
+    out_I_float = np_as_located_field(IDim)(np.zeros((size,), dtype=float64))
 
     @field_operator(backend=fieldview_backend)
     def unpack_tuple(
@@ -586,16 +585,14 @@ def test_solve_triag(fieldview_backend):
 
     solve_tridiag(a, b, c, d, out=out, offset_provider={})
 
-    np.allclose(expected, out)
+    assert np.allclose(expected, out)
 
 
-def test_ternary_operator(fieldview_backend):
+@pytest.mark.parametrize("left,right", [(2.0, 3.0), (3.0, 2.0)])
+def test_ternary_operator(left, right, fieldview_backend):
     a_I_float = np_as_located_field(IDim)(np.random.randn(size).astype("float64"))
     b_I_float = np_as_located_field(IDim)(np.random.randn(size).astype("float64"))
-    out_I_float = np_as_located_field(IDim)(np.zeros((size), dtype=float64))
-
-    left = 2.0
-    right = 3.0
+    out_I_float = np_as_located_field(IDim)(np.zeros((size,), dtype=float64))
 
     @field_operator(backend=fieldview_backend)
     def ternary_field_op(
@@ -605,27 +602,23 @@ def test_ternary_operator(fieldview_backend):
 
     ternary_field_op(a_I_float, b_I_float, left, right, out=out_I_float, offset_provider={})
     e = np.asarray(a_I_float) if left < right else np.asarray(b_I_float)
-    np.allclose(e, out_I_float)
+    assert np.allclose(e, out_I_float)
 
     @field_operator(backend=fieldview_backend)
     def ternary_field_op_scalars(left: float, right: float) -> Field[[IDim], float]:
         return broadcast(3.0, (IDim,)) if left > right else broadcast(4.0, (IDim,))
 
     ternary_field_op_scalars(left, right, out=out_I_float, offset_provider={})
-    e = np.full(e.shape, 3.0) if left > right else e
-    np.allclose(e, out_I_float)
+    e = np.full(e.shape, 3.0) if left > right else np.full(e.shape, 4.0)
+    assert np.allclose(e, out_I_float)
 
 
-def test_ternary_operator_tuple(fieldview_backend):
-    if fieldview_backend in [gtfn_cpu.run_gtfn, gtfn_cpu.run_gtfn_imperative]:
-        pytest.skip("Tuple arguments are not supported in gtfn yet.")
+@pytest.mark.parametrize("left,right", [(2.0, 3.0), (3.0, 2.0)])
+def test_ternary_operator_tuple(left, right, fieldview_backend):
     a_I_float = np_as_located_field(IDim)(np.random.randn(size).astype("float64"))
     b_I_float = np_as_located_field(IDim)(np.random.randn(size).astype("float64"))
-    out_I_float = np_as_located_field(IDim)(np.zeros((size), dtype=float64))
-    out_I_float_1 = np_as_located_field(IDim)(np.zeros((size), dtype=float64))
-
-    left = 2.0
-    right = 3.0
+    out_I_float = np_as_located_field(IDim)(np.zeros((size,), dtype=float64))
+    out_I_float_1 = np_as_located_field(IDim)(np.zeros((size,), dtype=float64))
 
     @field_operator(backend=fieldview_backend)
     def ternary_field_op(
@@ -642,8 +635,8 @@ def test_ternary_operator_tuple(fieldview_backend):
         if left < right
         else (np.asarray(b_I_float), np.asarray(a_I_float))
     )
-    np.allclose(e, out_I_float)
-    np.allclose(f, out_I_float_1)
+    assert np.allclose(e, out_I_float)
+    assert np.allclose(f, out_I_float_1)
 
 
 def test_ternary_builtin_neighbor_sum(reduction_setup, fieldview_backend):
@@ -694,9 +687,6 @@ def test_ternary_scan(fieldview_backend):
 
 @pytest.mark.parametrize("forward", [True, False])
 def test_scan_nested_tuple_output(fieldview_backend, forward):
-    if fieldview_backend in [gtfn_cpu.run_gtfn, gtfn_cpu.run_gtfn_imperative]:
-        pytest.xfail("gtfn does not yet support scan pass or tuple out arguments.")
-
     init = (1.0, (2.0, 3.0))
     out1, out2, out3 = (np_as_located_field(KDim)(np.zeros((size,))) for _ in range(3))
     expected = np.arange(1.0, 1.0 + size, 1)
@@ -718,11 +708,8 @@ def test_scan_nested_tuple_output(fieldview_backend, forward):
 
 @pytest.mark.parametrize("forward", [True, False])
 def test_scan_nested_tuple_input(fieldview_backend, forward):
-    if fieldview_backend in [gtfn_cpu.run_gtfn, gtfn_cpu.run_gtfn_imperative]:
-        pytest.xfail("gtfn does not yet support scan pass or tuple arguments.")
-
     init = 1.0
-    inp1 = np_as_located_field(KDim)(np.ones(size))
+    inp1 = np_as_located_field(KDim)(np.ones((size,)))
     inp2 = np_as_located_field(KDim)(np.arange(0.0, size, 1))
     out = np_as_located_field(KDim)(np.zeros((size,)))
 
@@ -760,58 +747,61 @@ def test_docstring():
 
 
 def test_domain(fieldview_backend):
-    a_IJ_float = np_as_located_field(IDim, JDim)(np.ones((size, size), dtype=float64))
+    inp = np_as_located_field(IDim, JDim)(np.ones((size, size), dtype=float64))
+    out = np_as_located_field(IDim, JDim)(2 * np.ones((size, size), dtype=float64))
+
+    expected = np.array(out)
+    expected[1:9, 4:6] = 1 + 1
 
     @field_operator(backend=fieldview_backend)
     def fieldop_domain(a: Field[[IDim, JDim], float64]) -> Field[[IDim, JDim], float64]:
         return a + a
 
     @program
-    def program_domain(a: Field[[IDim, JDim], float64]):
-        fieldop_domain(a, out=a, domain={IDim: (minimum(1, 2), 9), JDim: (4, maximum(5, 6))})
-
-    program_domain(a_IJ_float, offset_provider={})
+    def program_domain(inp: Field[[IDim, JDim], float64], out: Field[[IDim, JDim], float64]):
+        fieldop_domain(inp, out=out, domain={IDim: (minimum(1, 2), 9), JDim: (4, maximum(5, 6))})
 
-    expected = np.asarray(a_IJ_float)
-    expected[1:9, 4:6] = 1 + 1
+    program_domain(inp, out, offset_provider={})
 
-    assert np.allclose(expected, a_IJ_float)
+    assert np.allclose(expected, out)
 
 
 def test_domain_input_bounds(fieldview_backend):
     if fieldview_backend in [gtfn_cpu.run_gtfn, gtfn_cpu.run_gtfn_imperative]:
         pytest.skip("FloorDiv not fully supported in gtfn.")
-    a_IJ_float = np_as_located_field(IDim, JDim)(np.ones((size, size), dtype=float64))
+    inp = np_as_located_field(IDim, JDim)(np.ones((size, size), dtype=float64))
+    out = np_as_located_field(IDim, JDim)(2 * np.ones((size, size), dtype=float64))
 
     lower_i = 1
     upper_i = 9
     lower_j = 4
     upper_j = 6
 
+    expected = np.array(out)
+    expected[lower_i:upper_i, lower_j:upper_j] = 1 + 1
+
     @field_operator(backend=fieldview_backend)
     def fieldop_domain(a: Field[[IDim, JDim], float64]) -> Field[[IDim, JDim], float64]:
         return a + a
 
     @program
     def program_domain(
-        a: Field[[IDim, JDim], float64],
+        inp: Field[[IDim, JDim], float64],
+        out: Field[[IDim, JDim], float64],
         lower_i: int64,
         upper_i: int64,
         lower_j: int64,
         upper_j: int64,
     ):
         fieldop_domain(
-            a,
-            out=a,
+            inp,
+            out=out,
             domain={IDim: (lower_i, upper_i // 1), JDim: (lower_j**1, upper_j)},
         )
 
-    program_domain(a_IJ_float, lower_i, upper_i, lower_j, upper_j, offset_provider={})
-
-    expected = np.asarray(a_IJ_float)
-    expected[1:9, 4:6] = 1 + 1
+    program_domain(inp, out, lower_i, upper_i, lower_j, upper_j, offset_provider={})
 
-    assert np.allclose(expected, a_IJ_float)
+    assert np.allclose(expected, out)
 
 
 def test_domain_input_bounds_1(fieldview_backend):
@@ -822,9 +812,12 @@ def test_domain_input_bounds_1(fieldview_backend):
     lower_j = 4
     upper_j = 6
 
+    expected = np.array(a_IJ_float)
+    expected[lower_i:upper_i, lower_j:upper_j] = 1 + 1
+
     @field_operator(backend=fieldview_backend)
     def fieldop_domain(a: Field[[IDim, JDim], float64]) -> Field[[IDim, JDim], float64]:
-        return a * a
+        return a + a
 
     @program
     def program_domain(
@@ -842,33 +835,39 @@ def test_domain_input_bounds_1(fieldview_backend):
 
     program_domain(a_IJ_float, lower_i, upper_i, lower_j, upper_j, offset_provider={})
 
-    expected = np.asarray(a_IJ_float)
-    expected[1:9, 4:6] = 2 * 2
-
     assert np.allclose(expected, a_IJ_float)
 
 
 def test_domain_tuple(fieldview_backend):
-    a_IJ_float = np_as_located_field(IDim, JDim)(np.ones((size, size), dtype=float64))
-    b2d_float = a_IJ_float
+    inp0 = np_as_located_field(IDim, JDim)(np.ones((size, size), dtype=float64))
+    inp1 = np_as_located_field(IDim, JDim)(2 * np.ones((size, size), dtype=float64))
+    out0 = np_as_located_field(IDim, JDim)(3 * np.ones((size, size), dtype=float64))
+    out1 = np_as_located_field(IDim, JDim)(4 * np.ones((size, size), dtype=float64))
+
+    expected0 = np.array(out0)
+    expected0[1:9, 4:6] = (np.asarray(inp0) + np.asarray(inp1))[1:9, 4:6]
+    expected1 = np.array(out1)
+    expected1[1:9, 4:6] = np.asarray(inp1)[1:9, 4:6]
 
     @field_operator(backend=fieldview_backend)
     def fieldop_domain_tuple(
-        a: Field[[IDim, JDim], float64]
+        a: Field[[IDim, JDim], float64], b: Field[[IDim, JDim], float64]
     ) -> tuple[Field[[IDim, JDim], float64], Field[[IDim, JDim], float64]]:
-        return (a + a, a)
+        return (a + b, b)
 
     @program
-    def program_domain_tuple(a: Field[[IDim, JDim], float64], b: Field[[IDim, JDim], float64]):
-        fieldop_domain_tuple(a, out=(b, a), domain={IDim: (1, 9), JDim: (4, 6)})
-
-    program_domain_tuple(a_IJ_float, b2d_float, offset_provider={})
+    def program_domain_tuple(
+        inp0: Field[[IDim, JDim], float64],
+        inp1: Field[[IDim, JDim], float64],
+        out0: Field[[IDim, JDim], float64],
+        out1: Field[[IDim, JDim], float64],
+    ):
+        fieldop_domain_tuple(inp0, inp1, out=(out0, out1), domain={IDim: (1, 9), JDim: (4, 6)})
 
-    expected = np.asarray(a_IJ_float)
-    expected[1:9, 4:6] = 1 + 1
+    program_domain_tuple(inp0, inp1, out0, out1, offset_provider={})
 
-    assert np.allclose(np.asarray(a_IJ_float), a_IJ_float)
-    assert np.allclose(expected, b2d_float)
+    assert np.allclose(expected0, out0)
+    assert np.allclose(expected1, out1)
 
 
 def test_where_k_offset(fieldview_backend):
@@ -901,9 +900,6 @@ def test_undefined_symbols():
 
 
 def test_zero_dims_fields(fieldview_backend):
-    if fieldview_backend in [gtfn_cpu.run_gtfn, gtfn_cpu.run_gtfn_imperative]:
-        pytest.skip("Implicit broadcast are not supported yet.")
-
     inp = np_as_located_field()(np.array(1.0))
     out = np_as_located_field()(np.array(0.0))
 
@@ -916,9 +912,6 @@ def test_zero_dims_fields(fieldview_backend):
 
 
 def test_implicit_broadcast_mixed_dims(fieldview_backend):
-    if fieldview_backend == gtfn_cpu.run_gtfn:
-        pytest.skip("Implicit broadcast are not supported yet.")
-
     input1 = np_as_located_field(IDim)(np.ones((10,)))
     inp = np_as_located_field()(np.array(1.0))
     out = np_as_located_field(IDim)(np.ones((10,)))
@@ -939,15 +932,12 @@ def test_implicit_broadcast_mixed_dims(fieldview_backend):
 
 
 def test_tuple_unpacking(fieldview_backend):
-    if fieldview_backend in [gtfn_cpu.run_gtfn, gtfn_cpu.run_gtfn_imperative]:
-        pytest.skip("Tuple arguments are not supported in gtfn yet.")
-
     size = 10
-    inp = np_as_located_field(IDim)(np.ones((size)))
-    out1 = np_as_located_field(IDim)(np.ones((size)))
-    out2 = np_as_located_field(IDim)(np.ones((size)))
-    out3 = np_as_located_field(IDim)(np.ones((size)))
-    out4 = np_as_located_field(IDim)(np.ones((size)))
+    inp = np_as_located_field(IDim)(np.ones((size,)))
+    out1 = np_as_located_field(IDim)(np.ones((size,)))
+    out2 = np_as_located_field(IDim)(np.ones((size,)))
+    out3 = np_as_located_field(IDim)(np.ones((size,)))
+    out4 = np_as_located_field(IDim)(np.ones((size,)))
 
     @field_operator(backend=fieldview_backend)
     def unpack(
@@ -972,12 +962,9 @@ def test_tuple_unpacking(fieldview_backend):
 
 
 def test_tuple_unpacking_star_multi(fieldview_backend):
-    if fieldview_backend in [gtfn_cpu.run_gtfn, gtfn_cpu.run_gtfn_imperative]:
-        pytest.skip("Tuple arguments are not supported in gtfn yet.")
-
     size = 10
-    inp = np_as_located_field(IDim)(np.ones((size)))
-    out = tuple(np_as_located_field(IDim)(np.ones(size) * i) for i in range(3 * 4))
+    inp = np_as_located_field(IDim)(np.ones((size,)))
+    out = tuple(np_as_located_field(IDim)(np.ones((size,)) * i) for i in range(3 * 4))
 
     OutType = tuple[
         Field[[IDim], float64],
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/iterator_tests/test_tuple.py b/tests/next_tests/integration_tests/feature_tests/iterator_tests/test_tuple.py
index 84efbac9..44803f33 100644
--- a/tests/next_tests/integration_tests/feature_tests/iterator_tests/test_tuple.py
+++ b/tests/next_tests/integration_tests/feature_tests/iterator_tests/test_tuple.py
@@ -53,8 +53,8 @@ def tuple_output2(inp1, inp2):
     "stencil",
     [tuple_output1, tuple_output2],
 )
-def test_tuple_output(program_processor_no_gtfn_exec, stencil):
-    program_processor, validate = program_processor_no_gtfn_exec
+def test_tuple_output(program_processor, stencil):
+    program_processor, validate = program_processor
 
     shape = [5, 7, 9]
     rng = np.random.default_rng()
@@ -131,8 +131,8 @@ def test_tuple_of_field_of_tuple_output(program_processor_no_gtfn_exec):
         assert np.allclose(inp4, out_np2[:]["f1"])
 
 
-def test_tuple_of_tuple_of_field_output(program_processor_no_gtfn_exec):
-    program_processor, validate = program_processor_no_gtfn_exec
+def test_tuple_of_tuple_of_field_output(program_processor):
+    program_processor, validate = program_processor
 
     @fundef
     def stencil(inp1, inp2, inp3, inp4):
@@ -357,8 +357,8 @@ def tuple_input(inp):
     return tuple_get(0, inp_deref) + tuple_get(1, inp_deref)
 
 
-def test_tuple_field_input(program_processor_no_gtfn_exec):
-    program_processor, validate = program_processor_no_gtfn_exec
+def test_tuple_field_input(program_processor):
+    program_processor, validate = program_processor
 
     shape = [5, 7, 9]
     rng = np.random.default_rng()
@@ -475,8 +475,8 @@ def test_tuple_of_field_of_tuple_input(program_processor_no_gtfn_exec):
         assert np.allclose(2.0 * (np.asarray(inp1) + np.asarray(inp2)), out)
 
 
-def test_tuple_of_tuple_of_field_input(program_processor_no_gtfn_exec):
-    program_processor, validate = program_processor_no_gtfn_exec
+def test_tuple_of_tuple_of_field_input(program_processor):
+    program_processor, validate = program_processor
 
     shape = [5, 7, 9]
     rng = np.random.default_rng()
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 8281a850..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,21 +28,22 @@ 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()
         ),
     )
     compiled_program = build_the_program(example_program_source)
     buf = np.zeros(shape=(6, 5), dtype=np.float32)
+    tup = [np.zeros(shape=(6, 5), dtype=np.float32), np.zeros(shape=(6, 5), dtype=np.float32)]
     sc = np.float32(3.1415926)
-    res = compiled_program(buf, sc)
+    res = compiled_program(buf, tup, sc)
     assert math.isclose(res, 6 * 5 * 3.1415926, rel_tol=1e-4)
 
 
 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(),
@@ -50,6 +51,7 @@ def test_gtfn_cpp_with_compiledb(program_source_with_name):
     )
     compiled_program = build_the_program(example_program_source)
     buf = np.zeros(shape=(6, 5), dtype=np.float32)
+    tup = [np.zeros(shape=(6, 5), dtype=np.float32), np.zeros(shape=(6, 5), dtype=np.float32)]
     sc = np.float32(3.1415926)
-    res = compiled_program(buf, sc)
+    res = compiled_program(buf, tup, sc)
     assert math.isclose(res, 6 * 5 * 3.1415926, rel_tol=1e-4)
diff --git a/tests/next_tests/unit_tests/ffront_tests/test_foast_to_itir.py b/tests/next_tests/unit_tests/ffront_tests/test_foast_to_itir.py
index 6476c9fd..f11453c2 100644
--- a/tests/next_tests/unit_tests/ffront_tests/test_foast_to_itir.py
+++ b/tests/next_tests/unit_tests/ffront_tests/test_foast_to_itir.py
@@ -11,6 +11,11 @@
 # distribution for a copy of the license or check <https://www.gnu.org/licenses/>.
 #
 # SPDX-License-Identifier: GPL-3.0-or-later
+# TODO(tehrengruber): The style of the tests in this file is not optimal as a single change in the
+#  lowering can (and often does) make all of them fail. Once we have embedded field view we want to
+#  switch to executing the different cases here; once with a regular backend (i.e. including
+#  parsing) and then with embedded field view (i.e. no parsing). If the results match the lowering
+#  should be correct.
 
 from __future__ import annotations
 
diff --git a/tests/next_tests/unit_tests/iterator_tests/test_type_inference.py b/tests/next_tests/unit_tests/iterator_tests/test_type_inference.py
index 7f005df2..b5f40ae3 100644
--- a/tests/next_tests/unit_tests/iterator_tests/test_type_inference.py
+++ b/tests/next_tests/unit_tests/iterator_tests/test_type_inference.py
@@ -606,6 +606,21 @@ def test_function_definition():
     assert ti.pformat(inferred.dtype) == "(T₀) → T₀"
 
 
+def test_dynamic_offset():
+    """Test that the type of a dynamic offset is correctly inferred."""
+    offset_it = ir.SymRef(id="offset_it")
+    testee = ir.FunCall(
+        fun=ir.SymRef(id="shift"),
+        args=[
+            ir.OffsetLiteral(value="V2E"),
+            ir.FunCall(fun=ir.SymRef(id="deref"), args=[offset_it]),
+        ],
+    )
+    inferred_all: dict[int, ti.Type] = ti.infer_all(testee)
+    offset_it_type = inferred_all[id(offset_it)]
+    assert isinstance(offset_it_type, ti.Val) and offset_it_type.kind == ti.Iterator()
+
+
 CARTESIAN_DOMAIN = ir.FunCall(
     fun=ir.SymRef(id="cartesian_domain"),
     args=[
diff --git a/tests/next_tests/unit_tests/iterator_tests/transforms_tests/test_trace_shifts.py b/tests/next_tests/unit_tests/iterator_tests/transforms_tests/test_trace_shifts.py
index 4ad4d01a..8a5da3df 100644
--- a/tests/next_tests/unit_tests/iterator_tests/transforms_tests/test_trace_shifts.py
+++ b/tests/next_tests/unit_tests/iterator_tests/transforms_tests/test_trace_shifts.py
@@ -12,6 +12,7 @@
 #
 # SPDX-License-Identifier: GPL-3.0-or-later
 
+from gt4py.next.ffront import itir_makers as im
 from gt4py.next.iterator import ir
 from gt4py.next.iterator.transforms.trace_shifts import ALL_NEIGHBORS, TraceShifts
 
@@ -120,15 +121,52 @@ def test_neighbors():
 
 def test_reduce():
     testee = ir.StencilClosure(
-        stencil=ir.FunCall(
-            fun=ir.SymRef(id="reduce"),
-            args=[ir.SymRef(id="plus"), ir.SymRef(id="init")],
+        # λ(inp) → reduce(plus, init)(·inp)
+        stencil=ir.Lambda(
+            params=[ir.Sym(id="inp")],
+            expr=ir.FunCall(
+                fun=ir.FunCall(
+                    fun=ir.SymRef(id="reduce"), args=[ir.SymRef(id="plus"), ir.SymRef(id="init")]
+                ),
+                args=[ir.FunCall(fun=ir.SymRef(id="deref"), args=[ir.SymRef(id="inp")])],
+            ),
         ),
         inputs=[ir.SymRef(id="inp")],
         output=ir.SymRef(id="out"),
         domain=ir.FunCall(fun=ir.SymRef(id="cartesian_domain"), args=[]),
     )
-    expected = dict()
+    expected = {"inp": [()]}
+
+    actual = dict()
+    TraceShifts().visit(testee, shifts=actual)
+    assert actual == expected
+
+
+def test_shifted_literal():
+    "Test shifting an applied lift of a stencil returning a constant / literal works."
+    testee = ir.StencilClosure(
+        # λ(x) → ·⟪Iₒ, 1ₒ⟫((↑(λ() → 1))())
+        stencil=im.lambda__("x")(im.deref_(im.shift_("I", 1)(im.lift_(im.lambda__()(1))()))),
+        inputs=[ir.SymRef(id="inp")],
+        output=ir.SymRef(id="out"),
+        domain=ir.FunCall(fun=ir.SymRef(id="cartesian_domain"), args=[]),
+    )
+    expected = {"inp": []}
+
+    actual = dict()
+    TraceShifts().visit(testee, shifts=actual)
+    assert actual == expected
+
+
+def test_tuple_get_on_closure_input():
+    testee = ir.StencilClosure(
+        # λ(x) → (·⟪Iₒ, 1ₒ⟫(x))[0]
+        stencil=im.lambda__("x")(im.tuple_get_(0, im.deref_(im.shift_("I", 1)("x")))),
+        inputs=[ir.SymRef(id="inp")],
+        output=ir.SymRef(id="out"),
+        domain=ir.FunCall(fun=ir.SymRef(id="cartesian_domain"), args=[]),
+    )
+    expected = {"inp": [(ir.OffsetLiteral(value="I"), ir.OffsetLiteral(value=1))]}
 
     actual = dict()
     TraceShifts().visit(testee, shifts=actual)
diff --git a/tests/next_tests/unit_tests/otf_tests/binding_tests/test_cpp_interface.py b/tests/next_tests/unit_tests/otf_tests/binding_tests/test_cpp_interface.py
index fdfe10fc..1bed014a 100644
--- a/tests/next_tests/unit_tests/otf_tests/binding_tests/test_cpp_interface.py
+++ b/tests/next_tests/unit_tests/otf_tests/binding_tests/test_cpp_interface.py
@@ -107,3 +107,54 @@ def test_render_function_call_buffer(function_buffer_example):
     )
     expected = format_source("cpp", """example(get_arg_1(), get_arg_2())""", style="LLVM")
     assert rendered == expected
+
+
+@pytest.fixture
+def function_tuple_example():
+    return interface.Function(
+        name="example",
+        parameters=[
+            interface.Parameter(
+                name="a_buf",
+                type_=ts.TupleType(
+                    types=[
+                        ts.FieldType(
+                            dims=[Dimension("foo"), Dimension("bar")],
+                            dtype=ts.ScalarType(ts.ScalarKind.FLOAT64),
+                        ),
+                        ts.FieldType(
+                            dims=[Dimension("foo"), Dimension("bar")],
+                            dtype=ts.ScalarType(ts.ScalarKind.FLOAT64),
+                        ),
+                    ]
+                ),
+            )
+        ],
+    )
+
+
+def test_render_function_declaration_tuple(function_tuple_example):
+    rendered = format_source(
+        "cpp", cpp.render_function_declaration(function_tuple_example, "return;"), style="LLVM"
+    )
+    expected = format_source(
+        "cpp",
+        """\
+    template <class BufferT0>
+    decltype(auto) example(BufferT0&& a_buf) {
+        return;
+    }\
+    """,
+        style="LLVM",
+    )
+    assert rendered == expected
+
+
+def test_render_function_call_tuple(function_tuple_example):
+    rendered = format_source(
+        "cpp",
+        cpp.render_function_call(function_tuple_example, args=["get_arg_1()"]),
+        style="LLVM",
+    )
+    expected = format_source("cpp", """example(get_arg_1())""", style="LLVM")
+    assert rendered == expected
diff --git a/tests/next_tests/unit_tests/otf_tests/binding_tests/test_pybind.py b/tests/next_tests/unit_tests/otf_tests/binding_tests/test_pybind.py
index 1bd97c4b..df3de2fc 100644
--- a/tests/next_tests/unit_tests/otf_tests/binding_tests/test_pybind.py
+++ b/tests/next_tests/unit_tests/otf_tests/binding_tests/test_pybind.py
@@ -27,20 +27,36 @@ def test_bindings(program_source_example):
         #include "stencil.cpp.inc"
 
         #include <gridtools/common/defs.hpp>
+        #include <gridtools/common/tuple_util.hpp>
         #include <gridtools/fn/backend/naive.hpp>
         #include <gridtools/fn/cartesian.hpp>
         #include <gridtools/fn/unstructured.hpp>
+        #include <gridtools/sid/composite.hpp>
         #include <gridtools/sid/rename_dimensions.hpp>
         #include <gridtools/sid/unknown_kind.hpp>
         #include <gridtools/storage/adapter/python_sid_adapter.hpp>
         #include <pybind11/pybind11.h>
         #include <pybind11/stl.h>
 
-        decltype(auto) stencil_wrapper(pybind11::buffer buf, float sc) {
+        decltype(auto)
+        stencil_wrapper(pybind11::buffer buf,
+                        std::tuple<pybind11::buffer, pybind11::buffer> tup, float sc) {
           return stencil(
               gridtools::sid::rename_numbered_dimensions<generated::I_t,
-                                                         generated::J_t>(
-                  gridtools::as_sid<float, 2, gridtools::sid::unknown_kind>(buf)), sc);
+                                                        generated::J_t>(
+                  gridtools::as_sid<float, 2, gridtools::sid::unknown_kind>(buf)),
+              gridtools::sid::composite::keys<gridtools::integral_constant<int, 0>,
+                                              gridtools::integral_constant<int, 1>>::
+                  make_values(
+                      gridtools::sid::rename_numbered_dimensions<generated::I_t,
+                                                                generated::J_t>(
+                          gridtools::as_sid<
+                              float, 2, gridtools::sid::unknown_kind>(gridtools::tuple_util::get<0>(tup))),
+                      gridtools::sid::rename_numbered_dimensions<generated::I_t,
+                                                                generated::J_t>(
+                          gridtools::as_sid<
+                              float, 2, gridtools::sid::unknown_kind>(gridtools::tuple_util::get<1>(tup)))),
+              sc);
         }
 
         PYBIND11_MODULE(stencil, module) {
diff --git a/tests/next_tests/unit_tests/otf_tests/compilation_tests/build_systems_tests/conftest.py b/tests/next_tests/unit_tests/otf_tests/compilation_tests/build_systems_tests/conftest.py
index 443e193b..e5c2c17c 100644
--- a/tests/next_tests/unit_tests/otf_tests/compilation_tests/build_systems_tests/conftest.py
+++ b/tests/next_tests/unit_tests/otf_tests/compilation_tests/build_systems_tests/conftest.py
@@ -36,6 +36,21 @@ def make_program_source(name: str) -> stages.ProgramSource:
                     dtype=ts.ScalarType(ts.ScalarKind.FLOAT32),
                 ),
             ),
+            interface.Parameter(
+                name="tup",
+                type_=ts.TupleType(
+                    types=[
+                        ts.FieldType(
+                            dims=[common.Dimension("I"), common.Dimension("J")],
+                            dtype=ts.ScalarType(ts.ScalarKind.FLOAT32),
+                        ),
+                        ts.FieldType(
+                            dims=[common.Dimension("I"), common.Dimension("J")],
+                            dtype=ts.ScalarType(ts.ScalarKind.FLOAT32),
+                        ),
+                    ]
+                ),
+            ),
             interface.Parameter(name="sc", type_=ts.ScalarType(ts.ScalarKind.FLOAT32)),
         ],
     )
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]"