refactor: executor pipeline

marimo/_runtime/app/script_runner.py

@@ -19,11 +19,9 @@
 from marimo._runtime.exceptions import (
     MarimoMissingRefError,
     MarimoRuntimeException,
+    unwrap_user_exception,
 )
-from marimo._runtime.executor import (
-    ExecutionConfig,
-    get_executor,
-)
+from marimo._runtime.executor import resolve_executor
 from marimo._runtime.patches import (
     create_main_module,
     extract_docstring_from_header,
@@ -67,7 +65,7 @@ def __init__(
             if app.cell_manager.cell_data_at(cid).cell is not None
             and not self.app.graph.is_disabled(cid)
         )
-        self._executor = get_executor(ExecutionConfig())
+        self._executor = resolve_executor()
 
     def _cancel(self, cell_id: CellId_t) -> None:
         cancelled = {
@@ -108,15 +106,24 @@ def _run_synchronous(
                 cell = self.app.graph.cells[cid]
                 with get_context().with_cell_id(cid):
                     try:
-                        output = self._executor.execute_cell(
-                            cell, glbls, self.app.graph
-                        )
+                        output = self._executor.execute_cell(cell, glbls)
                         outputs[cid] = output
                     except MarimoRuntimeException as e:
-                        unwrapped_exception: BaseException | None = e.__cause__
+                        unwrapped_exception = unwrap_user_exception(
+                            e, self.app.graph
+                        )
 
                         if isinstance(unwrapped_exception, MarimoStopError):
                             self._cancel(cid)
+                        elif isinstance(
+                            unwrapped_exception, MarimoMissingRefError
+                        ):
+                            name_err = unwrapped_exception.name_error
+                            raise (
+                                name_err
+                                if name_err is not None
+                                else unwrapped_exception
+                            ) from None
                         else:
                             raise
                     finally:
@@ -155,14 +162,25 @@ async def _run_asynchronous(
                 with get_context().with_cell_id(cid):
                     try:
                         output = await self._executor.execute_cell_async(
-                            cell, glbls, self.app.graph
+                            cell, glbls
                         )
                         outputs[cid] = output
                     except MarimoRuntimeException as e:
-                        unwrapped_exception: BaseException | None = e.__cause__
+                        unwrapped_exception = unwrap_user_exception(
+                            e, self.app.graph
+                        )
 
                         if isinstance(unwrapped_exception, MarimoStopError):
                             self._cancel(cid)
+                        elif isinstance(
+                            unwrapped_exception, MarimoMissingRefError
+                        ):
+                            name_err = unwrapped_exception.name_error
+                            raise (
+                                name_err
+                                if name_err is not None
+                                else unwrapped_exception
+                            ) from None
                         else:
                             raise
                     finally:

marimo/_runtime/dataflow/runner.py

@@ -5,7 +5,7 @@
 
 from typing import TYPE_CHECKING, Any
 
-from marimo._runtime.executor import ExecutionConfig, get_executor
+from marimo._runtime.executor import DefaultExecutor
 
 if TYPE_CHECKING:
     from marimo._ast.cell import CellImpl
@@ -28,7 +28,7 @@ class Runner:
 
     def __init__(self, graph: DirectedGraph) -> None:
         self._graph = graph
-        self._executor = get_executor(ExecutionConfig())
+        self._executor = DefaultExecutor()
 
     @staticmethod
     def _returns(cell_impl: CellImpl, glbls: dict[str, Any]) -> dict[str, Any]:
@@ -99,13 +99,11 @@ async def run_cell_async(
 
         glbls: dict[str, Any] = {}
         for cid in topological_sort(graph, ancestor_ids):
-            await self._executor.execute_cell_async(
-                graph.cells[cid], glbls, graph
-            )
+            await self._executor.execute_cell_async(graph.cells[cid], glbls)
 
         Runner._substitute_refs(cell_impl, glbls, kwargs)
         output = await self._executor.execute_cell_async(
-            graph.cells[cell_impl.cell_id], glbls, graph
+            graph.cells[cell_impl.cell_id], glbls
         )
         defs = Runner._returns(cell_impl, glbls)
         return output, defs
@@ -142,11 +140,11 @@ def run_cell_sync(
 
         glbls: dict[str, Any] = {}
         for cid in topological_sort(graph, ancestor_ids):
-            self._executor.execute_cell(graph.cells[cid], glbls, graph)
+            self._executor.execute_cell(graph.cells[cid], glbls)
 
         self._substitute_refs(cell_impl, glbls, kwargs)
         output = self._executor.execute_cell(
-            graph.cells[cell_impl.cell_id], glbls, graph
+            graph.cells[cell_impl.cell_id], glbls
         )
         defs = Runner._returns(cell_impl, glbls)
         return output, defs

marimo/_runtime/exceptions.py

@@ -1,6 +1,11 @@
 # Copyright 2026 Marimo. All rights reserved.
 from __future__ import annotations
 
+from typing import TYPE_CHECKING
+
+if TYPE_CHECKING:
+    from marimo._runtime.dataflow import DirectedGraph
+
 
 class MarimoRuntimeException(BaseException):
     """Wrapper for all marimo runtime exceptions."""
@@ -19,3 +24,16 @@ def __init__(self, ref: str, name_error: NameError | None = None) -> None:
         super().__init__(ref)
         self.ref = ref
         self.name_error = name_error
+
+
+def unwrap_user_exception(
+    exc: MarimoRuntimeException,
+    graph: DirectedGraph | None = None,
+) -> BaseException | None:
+    """Extract the user exception from a ``MarimoRuntimeException``."""
+    cause = exc.__cause__
+    if graph is not None and isinstance(cause, NameError):
+        name = getattr(cause, "name", None)
+        if name and name in graph.definitions:
+            return MarimoMissingRefError(name, cause)
+    return cause