fix(runtime-sdk): Make NonRetryableError raised in a workflow step non-retryable

packages/runtime-sdk/src/workers/entrypoints.py

@@ -17,7 +17,11 @@
     python_from_rpc,
     python_to_rpc,
 )
-from .utils import _from_js_error, _is_js_instance
+from .utils import (
+    _from_js_error,
+    _is_js_instance,
+    import_sdk_javascript_module_async,
+)
 
 if TYPE_CHECKING:
     from js import DurableObjectState, Env, ExecutionContext
@@ -294,6 +298,22 @@ def wrapped_init(self, *args, **kwargs):
     return cls
 
 
+_workflows_js_module = None
+
+
+async def _wrap_js_workflow_step(js_step):
+    """
+    Wrap the JS `WorkflowStep` stub with `wrapWorkflowStep` from the SDK's `workflows.js`.
+
+    The wrapper makes a Python `NonRetryableError` raised inside a step reach the Workflows
+    engine as a JS error the engine recognises as non-retryable (see `workflows.js`).
+    """
+    global _workflows_js_module  # noqa: PLW0603
+    if _workflows_js_module is None:
+        _workflows_js_module = await import_sdk_javascript_module_async("workflows.js")
+    return _workflows_js_module.wrapWorkflowStep(js_step)
+
+
 def _wrap_workflow_step(cls):
     run_fn = cls.__dict__.get("run")
     if run_fn is None:
@@ -304,7 +324,7 @@ async def wrapped_run(self, event=None, step=None, /, *args, **kwargs):
         if event is not None:
             event = python_from_rpc(event)
         if step is not None:
-            step = _WorkflowStepWrapper(step)
+            step = _WorkflowStepWrapper(await _wrap_js_workflow_step(step))
 
         result = run_fn(self, event, step, *args, **kwargs)
 

packages/runtime-sdk/src/workers/utils.py

@@ -126,6 +126,25 @@ def import_from_javascript(module_name: str) -> Any:
         raise
 
 
+# Directory, relative to the worker bundle root, that `pywrangler sync` vendors Python packages
+# into. wrangler registers `.js`/`.mjs` files found under `python_modules/workers/` as ES modules
+# (everything else in `python_modules/` is opaque data), which is what lets them be imported via
+# `import_from_javascript`.
+_SDK_JS_MODULE_PREFIX = "python_modules/workers/"
+
+
+async def import_sdk_javascript_module_async(name: str) -> Any:
+    """
+    Asynchronous function to import an sdk js module
+
+    This does not rely on JSPI, so it also works with Pyodide 0.26.0a2.
+    """
+    try:
+        return await _pyodide_entrypoint_helper.doAnImport(_SDK_JS_MODULE_PREFIX + name)
+    except JsException as e:
+        raise ImportError(f"Failed to import '{name}': {e}") from e
+
+
 @contextmanager
 def patch_env(
     d: dict[str, Any] | Sequence[tuple[str, Any]] | None = None, **kwds: dict[str, Any]
@@ -144,15 +163,30 @@ def _to_python_exception(exc: JsException) -> Exception:
         return exc
 
 
+_NON_RETRYABLE_ERROR_NAME = "NonRetryableError"
+
+
 def _from_js_error(exc: JsException) -> Exception:
     # convert into Python exception after a full round trip
     # Python - JS - Python
     message = exc.message or ""
 
+    # A NonRetryableError raised inside a workflow step is translated by the
+    # runtime into a JS error named "NonRetryableError" before it reaches the
+    # Workflows engine, which is how the engine knows not to retry the step.
+    # When the engine rethrows it back to us the name either survives, or is
+    # folded into the message as a prefix.
+    if getattr(exc, "name", None) == _NON_RETRYABLE_ERROR_NAME:
+        return NonRetryableError(message)
+    if message == _NON_RETRYABLE_ERROR_NAME:
+        return NonRetryableError()
+    if message.startswith(_NON_RETRYABLE_ERROR_NAME + ": "):
+        return NonRetryableError(message[len(_NON_RETRYABLE_ERROR_NAME) + 2 :])
+
     # A Python exception that escaped to JS is a Pyodide `PythonError` whose
     # message is the formatted traceback. Depending on how it was serialized
     # over RPC it either keeps its name or arrives as a plain `Error` with
-    # "PythonError: " folded into the message.
+    # "PythonError: " folded into the message
     if getattr(exc, "name", None) != "PythonError" and not message.startswith(
         "PythonError"
     ):

packages/runtime-sdk/src/workers/workflows.js

@@ -0,0 +1,91 @@
+// JavaScript helpers for Python Workflows
+
+const NON_RETRYABLE_ERROR_NAME = "NonRetryableError";
+
+function isPythonError(e) {
+  return (
+    e instanceof Error &&
+    e.constructor?.name === "PythonError" &&
+    typeof e.type === "string"
+  );
+}
+
+// Extract the message the user passed to the Python exception from the
+// traceback stored in `PythonError.message`.
+//
+// The last non-empty line of a formatted traceback is
+// `<qualified.ExceptionType>: <message>`, or just `<qualified.ExceptionType>`
+// when the exception was raised without a message. For example, given
+//
+//   Traceback (most recent call last):
+//     File "/session/metadata/worker.py", line 19, in non_retryable
+//       raise NonRetryableError("do not retry")
+//   workers.workflows.NonRetryableError: do not retry
+//
+// this returns "do not retry", and given
+//
+//   Traceback (most recent call last):
+//     File "/session/metadata/worker.py", line 34, in no_message
+//       raise NonRetryableError()
+//   workers.workflows.NonRetryableError
+//
+// it returns "".
+function pythonExceptionMessage(e) {
+  const lines = e.message.split("\n").filter((line) => line.trim() !== "");
+  const last = lines.at(-1) ?? "";
+  const sep = last.indexOf(": ");
+  return sep === -1 ? "" : last.slice(sep + 2);
+}
+
+// Wraps a Python step callback passed to `WorkflowStep.do()`.
+//
+// If the error thrown is a Python NonRetryable error, translate it into a JS
+// error that the engine will recognize as non retryable. Leave other errors
+// alone.
+export function wrapWorkflowStepCallback(pyCallback) {
+  return async function (...args) {
+    try {
+      return await pyCallback(...args);
+    } catch (e) {
+      if (isPythonError(e) && e.type === NON_RETRYABLE_ERROR_NAME) {
+        const err = new Error(pythonExceptionMessage(e));
+        err.name = NON_RETRYABLE_ERROR_NAME;
+        throw err;
+      }
+      throw e;
+    }
+  };
+}
+
+// Wraps the `WorkflowStep` RPC stub passed to a Python
+// `WorkflowEntrypoint.run()` so that the callback given to `step.do()` goes
+// through `wrapWorkflowStepCallback`.
+//
+// Because the wrapped `do` returns the stub's own promise, a Python callback
+// passed to it keeps exactly the lifetime it would have had without the
+// wrapper.
+export function wrapWorkflowStep(step) {
+  // RPC stubs are callable, so `typeof step` is 'function'.
+  if (
+    step === null ||
+    (typeof step !== "object" && typeof step !== "function")
+  ) {
+    return step;
+  }
+  return new Proxy(step, {
+    apply(target, thisArg, args) {
+      return Reflect.apply(target, thisArg, args);
+    },
+    get(target, prop) {
+      if (prop !== "do") {
+        return Reflect.get(target, prop);
+      }
+      return function (name, ...rest) {
+        const args = rest.map((arg) =>
+          typeof arg === "function" ? wrapWorkflowStepCallback(arg) : arg
+        );
+        return target.do(name, ...args);
+      };
+    },
+  });
+}

packages/runtime-sdk/tests/bindings-test/src/test_workflow.py

@@ -149,7 +149,17 @@ async def test_step_retry_config(env):
 async def test_non_retryable_error(env):
     instance = await env.MY_WORKFLOW.create({"params": {"mode": "non_retryable"}})
     status = await _poll(instance)
-    assert status["status"] == "errored", f"unexpected status: {dict(status)!r}"
+    assert status["status"] == "complete", f"unexpected status: {dict(status)!r}"
+    out = status["output"]
+    # The step would succeed on attempt 2, so `retried` means the engine ignored
+    # the NonRetryableError.
+    assert out["retried"] is False, f"step was retried: {out!r}"
+    # ...and the error must come back to `run()` as a Python NonRetryableError.
+    assert out["caught"] == "NonRetryableError", out
+    # The message is preserved end to end in workerd, but miniflare currently
+    # truncates error messages crossing from the Python worker into its
+    # Workflows engine at the first ": " so accept an empty message here.
+    assert out["message"] in ("do not retry", ""), out
 
 
 async def test_error_handling_catch(env):

packages/runtime-sdk/tests/bindings-test/src/worker_workflow.py

@@ -109,14 +109,30 @@ async def flaky(ctx):
         return await flaky()
 
     async def _non_retryable(self, event, step):
+        # Fails on the first attempt with NonRetryableError and would succeed on a
+        # second attempt. If the engine honours NonRetryableError the step fails
+        # once and `run()` sees a NonRetryableError; if the engine retries, the step
+        # completes with `attempt == 2`.
         @step.do(
             "non-retryable-step",
-            config={"retries": {"limit": 1, "delay": 0, "backoff": "constant"}},
+            config={"retries": {"limit": 3, "delay": 0, "backoff": "constant"}},
         )
-        async def boom():
-            raise NonRetryableError("do not retry")
+        async def boom(ctx):
+            if int(ctx["attempt"]) < 2:
+                raise NonRetryableError("do not retry")
+            return {"retried": True, "attempt": int(ctx["attempt"])}
 
-        return await boom()
+        try:
+            result = await boom()
+        except NonRetryableError as exc:
+            return {
+                "retried": False,
+                "caught": "NonRetryableError",
+                "message": str(exc),
+            }
+        except Exception as exc:
+            return {"retried": False, "caught": type(exc).__name__, "message": str(exc)}
+        return result
 
     async def _duplicate_step_names(self, event, step):
         # The engine disambiguates repeated step names with a counter, so two

packages/runtime-sdk/tests/test_in_workerd.py

@@ -66,14 +66,27 @@ def embed(dir: Path, root: Path, level: int = 0):
         module_path = path.absolute().relative_to(module_path_root)
         embed_path = path.absolute().relative_to(root)
         if path.suffix == ".py":
-            modules.append(
-                f'(name = "{module_path}", pythonModule = embed "{embed_path}")'
-            )
+            module_type = "pythonModule"
+        elif _is_sdk_js_module(path.relative_to(dir)):
+            module_type = "esModule"
         else:
-            modules.append(f'(name = "{module_path}", data = embed "{embed_path}")')
+            module_type = "data"
+        modules.append(
+            f'(name = "{module_path}", {module_type} = embed "{embed_path}")'
+        )
     return modules
 
 
+def _is_sdk_js_module(path_in_vendor_dir: Path) -> bool:
+    """Mirror wrangler: `.js`/`.mjs` files under `python_modules/workers/` are ES modules.
+
+    Other vendored packages may ship `.js` assets that are not valid ES modules, so wrangler
+    only applies this rule to the SDK's own `workers/` package. Everything else stays `data`.
+    """
+    in_sdk_package = path_in_vendor_dir.parts[:1] == ("workers",)
+    return in_sdk_package and path_in_vendor_dir.suffix in (".js", ".mjs")
+
+
 @pytest.fixture(scope="module")
 def bundle_cache_dir(tmp_path_factory):
     yield tmp_path_factory.mktemp("bundle_cache")

packages/runtime-sdk/tests/workerd-test/workflow/pyproject.toml

@@ -0,0 +1,5 @@
+[project]
+name = "test"
+version = "0.0.0"
+requires-python = ">=3.12"
+dependencies = []

packages/runtime-sdk/tests/workerd-test/workflow/worker.js

@@ -0,0 +1,112 @@
+// Copyright (c) 2026 Cloudflare, Inc.
+// Licensed under the Apache 2.0 license found in the LICENSE file or at:
+//     https://opensource.org/licenses/Apache-2.0
+import { RpcTarget } from 'cloudflare:workers';
+
+import * as assert from 'node:assert';
+
+// A stand-in for the Workflows engine's `WorkflowStep`. Like the real engine it runs the step
+// callback over RPC, and it records the error it receives when the callback fails so the test
+// can assert on what the engine would have seen.
+class Context extends RpcTarget {
+  constructor() {
+    super();
+    this.errors = [];
+    this.sleeps = [];
+  }
+
+  async do(name, ...rest) {
+    // do(name, callback) or do(name, config, callback)
+    const callback = rest.at(-1);
+    try {
+      return await callback({
+        step: { name, count: 1 },
+        attempt: 1,
+        config: {},
+      });
+    } catch (e) {
+      this.errors.push({ step: name, name: e.name, message: e.message });
+      // The engine rethrows so the workflow's `run()` can handle the error.
+      throw e;
+    }
+  }
+
+  async sleep(name, duration) {
+    this.sleeps.push({ name, duration });
+  }
+}
+
+// The engine treats a step error as non-retryable when
+// `error.name === 'NonRetryableError' || error.message.startsWith('NonRetryableError')`
+// (the name is folded into the message when the error is tunneled over RPC without enhanced
+// error serialization).
+function isNonRetryable(e) {
+  return (
+    e.name === 'NonRetryableError' || e.message.startsWith('NonRetryableError')
+  );
+}
+
+function nonRetryableMessage(e) {
+  return e.name === 'NonRetryableError'
+    ? e.message
+    : e.message.replace(/^NonRetryableError(: )?/, '');
+}
+
+export default {
+  async test(ctrl, env) {
+    const step = new Context();
+    const result = await env.PythonWorkflow.run(
+      { payload: { foo: 'bar' } },
+      step
+    );
+
+    // --- What the engine saw -------------------------------------------------------------
+    const byStep = Object.fromEntries(step.errors.map((e) => [e.step, e]));
+    assert.deepStrictEqual(Object.keys(byStep).sort(), [
+      'non_retryable',
+      'non_retryable_no_message',
+      'type_error',
+    ]);
+
+    // A Python NonRetryableError must arrive as a non-retryable JS error carrying the
+    // Python message, not as a generic PythonError with a traceback.
+    assert.ok(isNonRetryable(byStep.non_retryable), byStep.non_retryable);
+    assert.strictEqual(nonRetryableMessage(byStep.non_retryable), 'do not retry');
+    assert.ok(
+      isNonRetryable(byStep.non_retryable_no_message),
+      byStep.non_retryable_no_message
+    );
+    assert.strictEqual(
+      nonRetryableMessage(byStep.non_retryable_no_message),
+      ''
+    );
+
+    // Other Python exceptions are left alone: they arrive as Pyodide's PythonError with the
+    // traceback as the message. Depending on the compatibility date the error is serialized
+    // either as `{name: 'PythonError', message: '<traceback>'}` (enhanced error
+    // serialization) or as `{name: 'Error', message: 'PythonError: <traceback>'}`.
+    assert.ok(!isNonRetryable(byStep.type_error), byStep.type_error);
+    assert.match(
+      `${byStep.type_error.name}: ${byStep.type_error.message}`,
+      /PythonError: Traceback/
+    );
+    assert.match(byStep.type_error.message, /TypeError: intentional type error/);
+
+    // --- What the Python workflow saw once the engine rethrew ----------------------------
+    assert.deepStrictEqual(result.non_retryable, {
+      caught: 'NonRetryableError',
+      message: 'do not retry',
+    });
+    assert.deepStrictEqual(result.non_retryable_no_message, {
+      caught: 'NonRetryableError',
+      message: '',
+    });
+    assert.strictEqual(result.type_error.caught, 'TypeError');
+    assert.match(result.type_error.message, /intentional type error/);
+
+    // The wrapped `step` still behaves like the real one otherwise.
+    assert.deepStrictEqual(result.ok, { attempt: 1, payload: { foo: 'bar' } });
+    assert.strictEqual(result.slept, true);
+    assert.deepStrictEqual(step.sleeps, [{ name: 'nap', duration: '1 second' }]);
+  },
+};