From 469d47c15b93dff5880fe295620600eaff907bf4 Mon Sep 17 00:00:00 2001 From: Hood Chatham Date: Wed, 16 Sep 2026 18:49:29 -0700 Subject: [PATCH 1/2] feat(runtime-sdk): Add helpers to import JavaScript modules shipped with the SDK wrangler registers `.js`/`.mjs` files under `python_modules/workers/` as ES modules, so JavaScript can ship inside the `workers` package and be loaded at runtime. --- packages/runtime-sdk/src/workers/utils.py | 19 +++++++++++++++++ packages/runtime-sdk/tests/test_in_workerd.py | 21 +++++++++++++++---- 2 files changed, 36 insertions(+), 4 deletions(-) diff --git a/packages/runtime-sdk/src/workers/utils.py b/packages/runtime-sdk/src/workers/utils.py index 07a2e985..b79e89db 100644 --- a/packages/runtime-sdk/src/workers/utils.py +++ b/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] diff --git a/packages/runtime-sdk/tests/test_in_workerd.py b/packages/runtime-sdk/tests/test_in_workerd.py index cbfce3b7..4f63fe3c 100644 --- a/packages/runtime-sdk/tests/test_in_workerd.py +++ b/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") From 2ccc8643cc1edb423caecb5aa00c3875f5232d2f Mon Sep 17 00:00:00 2001 From: Hood Chatham Date: Wed, 16 Sep 2026 18:49:29 -0700 Subject: [PATCH 2/2] fix(runtime-sdk): Make NonRetryableError raised in a workflow step non-retryable The Workflows engine decides whether to retry a failed step from the JS error's name. A Python exception reaches it as a Pyodide PythonError whose message is the traceback, so `workers.workflows.NonRetryableError` was silently retried like any other error. This adds a new workflows.js shim that fixes the translation of NonRetryableError to JavaScript so this works correctly. --- .../runtime-sdk/src/workers/entrypoints.py | 24 +++- packages/runtime-sdk/src/workers/utils.py | 17 ++- packages/runtime-sdk/src/workers/workflows.js | 91 ++++++++++++++ .../tests/bindings-test/src/test_workflow.py | 12 +- .../bindings-test/src/worker_workflow.py | 24 +++- .../workerd-test/workflow/pyproject.toml | 5 + .../tests/workerd-test/workflow/worker.js | 112 ++++++++++++++++++ .../tests/workerd-test/workflow/worker.py | 72 +++++++++++ .../workerd-test/workflow/workflow.wd-test | 30 +++++ .../workerd-test/workflow/wrangler.jsonc | 5 + 10 files changed, 384 insertions(+), 8 deletions(-) create mode 100644 packages/runtime-sdk/src/workers/workflows.js create mode 100644 packages/runtime-sdk/tests/workerd-test/workflow/pyproject.toml create mode 100644 packages/runtime-sdk/tests/workerd-test/workflow/worker.js create mode 100644 packages/runtime-sdk/tests/workerd-test/workflow/worker.py create mode 100644 packages/runtime-sdk/tests/workerd-test/workflow/workflow.wd-test create mode 100644 packages/runtime-sdk/tests/workerd-test/workflow/wrangler.jsonc diff --git a/packages/runtime-sdk/src/workers/entrypoints.py b/packages/runtime-sdk/src/workers/entrypoints.py index 058f357a..fe56a427 100644 --- a/packages/runtime-sdk/src/workers/entrypoints.py +++ b/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) diff --git a/packages/runtime-sdk/src/workers/utils.py b/packages/runtime-sdk/src/workers/utils.py index b79e89db..8ce5ae0d 100644 --- a/packages/runtime-sdk/src/workers/utils.py +++ b/packages/runtime-sdk/src/workers/utils.py @@ -163,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" ): diff --git a/packages/runtime-sdk/src/workers/workflows.js b/packages/runtime-sdk/src/workers/workflows.js new file mode 100644 index 00000000..37888e9d --- /dev/null +++ b/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 +// `: `, or just `` +// 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); + }; + }, + }); +} diff --git a/packages/runtime-sdk/tests/bindings-test/src/test_workflow.py b/packages/runtime-sdk/tests/bindings-test/src/test_workflow.py index 42135934..8feb8f21 100644 --- a/packages/runtime-sdk/tests/bindings-test/src/test_workflow.py +++ b/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): diff --git a/packages/runtime-sdk/tests/bindings-test/src/worker_workflow.py b/packages/runtime-sdk/tests/bindings-test/src/worker_workflow.py index 674ddd07..74b42b5b 100644 --- a/packages/runtime-sdk/tests/bindings-test/src/worker_workflow.py +++ b/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 diff --git a/packages/runtime-sdk/tests/workerd-test/workflow/pyproject.toml b/packages/runtime-sdk/tests/workerd-test/workflow/pyproject.toml new file mode 100644 index 00000000..072f326e --- /dev/null +++ b/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 = [] diff --git a/packages/runtime-sdk/tests/workerd-test/workflow/worker.js b/packages/runtime-sdk/tests/workerd-test/workflow/worker.js new file mode 100644 index 00000000..dd138764 --- /dev/null +++ b/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: ''}` (enhanced error + // serialization) or as `{name: 'Error', message: 'PythonError: '}`. + 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' }]); + }, +}; diff --git a/packages/runtime-sdk/tests/workerd-test/workflow/worker.py b/packages/runtime-sdk/tests/workerd-test/workflow/worker.py new file mode 100644 index 00000000..1f6173c9 --- /dev/null +++ b/packages/runtime-sdk/tests/workerd-test/workflow/worker.py @@ -0,0 +1,72 @@ +from workers import WorkflowEntrypoint +from workers.workflows import NonRetryableError + + +class TestWorkflow(WorkflowEntrypoint): + """ + Exercises the error path between Python steps and the Workflows engine. + + The JS side (worker.js) plays the engine and records the errors each step callback + throws; the results returned from here record what `run()` sees once the engine rethrows. + """ + + async def run(self, event, step): + results = {} + + @step.do("non_retryable") + async def non_retryable(): + raise NonRetryableError("do not retry") + + try: + await non_retryable() + except NonRetryableError as e: + results["non_retryable"] = { + "caught": "NonRetryableError", + "message": str(e), + } + except Exception as e: # pragma: no cover - reported to the JS side + results["non_retryable"] = {"caught": type(e).__name__, "message": str(e)} + + @step.do("non_retryable_no_message") + async def non_retryable_no_message(): + raise NonRetryableError() + + try: + await non_retryable_no_message() + except NonRetryableError as e: + results["non_retryable_no_message"] = { + "caught": "NonRetryableError", + "message": str(e), + } + except Exception as e: # pragma: no cover - reported to the JS side + results["non_retryable_no_message"] = { + "caught": type(e).__name__, + "message": str(e), + } + + # Ordinary Python exceptions are not translated: they reach the engine as a + # PythonError and come back with their type recovered from the traceback. + @step.do("type_error") + async def type_error(): + raise TypeError("intentional type error") + + try: + await type_error() + except TypeError as e: + results["type_error"] = {"caught": "TypeError", "message": str(e)} + except Exception as e: # pragma: no cover - reported to the JS side + results["type_error"] = {"caught": type(e).__name__, "message": str(e)} + + # A successful step still works through the wrapped `step`, including receiving + # the step context and the workflow event. + @step.do("ok") + async def ok(ctx): + return {"attempt": ctx["attempt"], "payload": event["payload"]} + + results["ok"] = await ok() + + # Steps using the other `step` methods are forwarded to the engine untouched. + await step.sleep("nap", "1 second") + results["slept"] = True + + return results diff --git a/packages/runtime-sdk/tests/workerd-test/workflow/workflow.wd-test b/packages/runtime-sdk/tests/workerd-test/workflow/workflow.wd-test new file mode 100644 index 00000000..6a46f791 --- /dev/null +++ b/packages/runtime-sdk/tests/workerd-test/workflow/workflow.wd-test @@ -0,0 +1,30 @@ +using Workerd = import "/workerd/workerd.capnp"; + +const config :Workerd.Config = ( + services = [ + (name = "py", worker = .pyWorker), + (name = "js", worker = .jsWorker), + ], +); + +const pyWorker :Workerd.Worker = ( + modules = [ + (name = "worker.py", pythonModule = embed "worker.py"), + %PYTHON_MODULES + ], + compatibilityDate = "%COMPAT_DATE", + compatibilityFlags = ["python_workers", "python_workflows", "python_workflows_implicit_dependencies"], +); + +# Stands in for the Workflows engine: calls the Python WorkflowEntrypoint over RPC and hands it +# a `step` RpcTarget, exactly like the real engine does. +const jsWorker :Workerd.Worker = ( + compatibilityFlags = ["nodejs_compat"], + compatibilityDate = "%COMPAT_DATE", + modules = [ + (name = "worker", esModule = embed "worker.js"), + ], + bindings = [ + (name = "PythonWorkflow", service = (name = "py", entrypoint = "TestWorkflow")), + ], +); diff --git a/packages/runtime-sdk/tests/workerd-test/workflow/wrangler.jsonc b/packages/runtime-sdk/tests/workerd-test/workflow/wrangler.jsonc new file mode 100644 index 00000000..9b5a9609 --- /dev/null +++ b/packages/runtime-sdk/tests/workerd-test/workflow/wrangler.jsonc @@ -0,0 +1,5 @@ +{ + "name": "test-worker", + "compatibility_date": "%COMPAT_DATE", + "compatibility_flags": ["python_workers", "python_workflows", "python_workflows_implicit_dependencies"] +}