Various workflows bug fixes

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
 

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"):

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.

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",

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"]