diff --git a/packages/runtime-sdk/src/workers/entrypoints.py b/packages/runtime-sdk/src/workers/entrypoints.py index 50131e13..058f357a 100644 --- a/packages/runtime-sdk/src/workers/entrypoints.py +++ b/packages/runtime-sdk/src/workers/entrypoints.py @@ -116,7 +116,7 @@ async def wrapper(): func, depends, implicit ) results = await self._gather_results(results_future_list, concurrent) - return await _do_call(self, step_name, config, func, *results) + return await _do_call(self, step_name, wrapper, config, func, *results) wrapper._step_name = step_name self.step_closures[step_name] = wrapper @@ -157,7 +157,13 @@ def _build_dependency_list(self, func, depends, implicit): if p.name == "ctx": results_future_list.append(p) else: - results_future_list.append(depends[curr]) + dep = depends[curr] + if not hasattr(dep, "_step_name"): + raise TypeError( + f"'depends' entry for parameter {p.name!r} is not a " + "function decorated with step.do" + ) + results_future_list.append(dep) curr += 1 return results_future_list @@ -194,17 +200,17 @@ async def wait_for_event(self, name, event_type, /, timeout="24 hours"): ) async def _resolve_dependency(self, dep): - if hasattr(dep, "name") and dep.name == "ctx": + if isinstance(dep, inspect.Parameter) and dep.name == "ctx": return dep - elif dep._step_name in self._memoized_dependencies: - return self._memoized_dependencies[dep._step_name] - elif dep._step_name in self._in_flight: - return await self._in_flight[dep._step_name] + elif dep in self._memoized_dependencies: + return self._memoized_dependencies[dep] + elif dep in self._in_flight: + return await self._in_flight[dep] return await dep() -async def _do_call(entrypoint, name, config, callback, *results): +async def _do_call(entrypoint, name, key, config, callback, *results): async def _callback(ctx=None): # deconstruct the actual ctx object resolved_results = tuple( @@ -217,7 +223,8 @@ async def _callback(ctx=None): if inspect.iscoroutine(result): result = await result - return to_js(result, dict_converter=Object.fromEntries) + # The step result crosses the RPC boundary back to the Workflows engine + return python_to_rpc(result) async def _closure(): try: @@ -233,13 +240,13 @@ async def _closure(): raise _from_js_error(exc) from exc task = create_task(_closure()) - entrypoint._in_flight[name] = task + entrypoint._in_flight[key] = task try: result = await task - entrypoint._memoized_dependencies[name] = result + entrypoint._memoized_dependencies[key] = result finally: - del entrypoint._in_flight[name] + entrypoint._in_flight.pop(key, None) return result diff --git a/packages/runtime-sdk/src/workers/utils.py b/packages/runtime-sdk/src/workers/utils.py index ea4018d2..07a2e985 100644 --- a/packages/runtime-sdk/src/workers/utils.py +++ b/packages/runtime-sdk/src/workers/utils.py @@ -147,11 +147,24 @@ def _to_python_exception(exc: JsException) -> Exception: def _from_js_error(exc: JsException) -> Exception: # convert into Python exception after a full round trip # Python - JS - Python - if not exc.message or not exc.message.startswith("PythonError"): + message = exc.message or "" + + # 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. + if getattr(exc, "name", None) != "PythonError" and not message.startswith( + "PythonError" + ): return _to_python_exception(exc) - # extract the Python exception type from the traceback - error_message_last_line = exc.message.split("\n")[-2] + # extract the Python exception type from the last line of the traceback. The + # message may have been stripped down to just "PythonError" when crossing an RPC + # boundary, in which case there is no traceback to inspect. + lines = message.rstrip().split("\n") + if len(lines) < 2: + return _to_python_exception(exc) + error_message_last_line = lines[-1] if error_message_last_line.startswith("TypeError"): return TypeError(error_message_last_line) elif error_message_last_line.startswith("ValueError"): 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 85fd0c6b..42135934 100644 --- a/packages/runtime-sdk/tests/bindings-test/src/test_workflow.py +++ b/packages/runtime-sdk/tests/bindings-test/src/test_workflow.py @@ -159,11 +159,36 @@ async def test_error_handling_catch(env): # Per the docs, a step error propagates to run() and is catchable with # `except Exception`. Neither the concrete type nor the original message is # guaranteed to survive the RPC layer, so we assert the reliable contract: - # the error was caught and a message was produced. + # the error was caught and a message was produced, and that the SDK's own + # error translation did not blow up (e.g. IndexError while parsing it). assert status["output"]["caught"] is not None + assert status["output"]["caught"] != "IndexError" assert status["output"]["message"] +async def test_duplicate_step_names(env): + instance = await env.MY_WORKFLOW.create( + {"params": {"mode": "duplicate_step_names"}} + ) + status = await _poll(instance) + assert status["status"] == "complete", f"unexpected status: {dict(status)!r}" + assert status["output"]["concurrent"] == [1, 2] + assert status["output"]["uses"] == 20 + + +async def test_step_output_conversion(env): + instance = await env.MY_WORKFLOW.create( + {"params": {"mode": "step_output_conversion"}} + ) + status = await _poll(instance) + assert status["status"] == "complete", f"unexpected status: {dict(status)!r}" + out = status["output"] + assert out["when_is_datetime"] is True + assert out["year"] == 2026 + assert out["nothing_is_none"] is True + assert out["nested_nothing_is_none"] is True + + # The tests below pass pre-converted (to_js) objects, the legacy pattern from the # Workflows docs, to ensure existing code keeps working now that the binding # auto-converts plain Python objects. 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 2dc163c2..674ddd07 100644 --- a/packages/runtime-sdk/tests/bindings-test/src/worker_workflow.py +++ b/packages/runtime-sdk/tests/bindings-test/src/worker_workflow.py @@ -1,3 +1,4 @@ +import asyncio import datetime from workers import WorkflowEntrypoint @@ -19,6 +20,8 @@ async def run(self, event, step): "retry": self._retry, "non_retryable": self._non_retryable, "catch_error": self._catch_error, + "duplicate_step_names": self._duplicate_step_names, + "step_output_conversion": self._step_output_conversion, } handler = handlers.get(mode) if handler is None: @@ -115,6 +118,47 @@ async def boom(): return await boom() + async def _duplicate_step_names(self, event, step): + # The engine disambiguates repeated step names with a counter, so two + # steps sharing a name must not share memoised results or in-flight tasks. + @step.do("dup") + async def first(): + return 1 + + @step.do("dup") + async def second(): + return 2 + + # Implicit dependencies resolve by step name, so `dup` refers to the most + # recently registered closure (`second`) and must return its result. + @step.do() + async def uses(dup): + return dup * 10 + + # Run both same-named steps concurrently so they are in flight together. + concurrent = list(await asyncio.gather(first(), second())) + return {"concurrent": concurrent, "uses": await uses()} + + async def _step_output_conversion(self, event, step): + @step.do("produce") + async def produce(): + return { + "when": datetime.datetime(2026, 1, 2, 3, 4, 5), + "nothing": None, + "nested": {"nothing": None}, + } + + @step.do() + async def consume(produce): + return { + "when_is_datetime": isinstance(produce["when"], datetime.datetime), + "year": produce["when"].year, + "nothing_is_none": produce["nothing"] is None, + "nested_nothing_is_none": produce["nested"]["nothing"] is None, + } + + return await consume() + async def _catch_error(self, event, step): @step.do( "failing-step", diff --git a/packages/testlib/testlib/entrypoint.py b/packages/testlib/testlib/entrypoint.py index 4df2bb55..4d4c6bd8 100644 --- a/packages/testlib/testlib/entrypoint.py +++ b/packages/testlib/testlib/entrypoint.py @@ -59,7 +59,8 @@ def pytest_runtest_makereport(self, item, call): if report.when == "call": excinfo = call.excinfo if excinfo is None: - result["error"] = "unknown error" + # e.g. a strict xfail that unexpectedly passed. + result["error"] = report.longreprtext or "unknown error" elif excinfo.errisinstance(AssertionError): result["error"] = str(excinfo.value) del result["traceback"]