Fix async callable wrapper traversal

Track identities across wrapped, partial, and callable nodes; recognize intermediate coroutine functions before following their wrapped targets. Cover bounded worker failures and nested decorated dispatch.

Co-authored-by: Copilot App <223556219+Copilot@users.noreply.github.com>

Author: andystaples

CHANGELOG.md

@@ -9,6 +9,10 @@ adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0.html).
 
 FIXED
 
+- Async orchestrators wrapped with `functools.wraps` are now recognized even
+when an intermediate async wrapper decorates a synchronous function, including
+callable instances and partials. Cyclic wrapper chains fail instead of hanging
+worker dispatch.
 - Activity and sub-orchestration `retry_timeout` decisions now use replay-safe
 orchestration time, preventing retries from expiring solely because history is
 replayed later. Retry calculations also accept timezone-aware timestamps.

durabletask/internal/shared.py

@@ -76,19 +76,27 @@ def from_json(json_str: str | bytes | bytearray, expected_type: type | None = No
 
 def is_async_callable(fn: Any) -> bool:
     """Recognize async functions, wrapped functions, partials and callable objects."""
+    # Keep nodes alive so temporary bound-method IDs cannot be reused.
+    seen: dict[int, Any] = {}
     while True:
+        identity = id(fn)
+        if identity in seen:
+            raise ValueError("wrapper loop when classifying async callable")
+        seen[identity] = fn
         if inspect.iscoroutinefunction(fn):
             return True
-        fn = inspect.unwrap(fn)
+        if hasattr(fn, "__wrapped__"):
+            fn = fn.__wrapped__
+            continue
         if isinstance(fn, partial):
             fn = fn.func
             continue
-        if inspect.iscoroutinefunction(fn):
-            return True
-        if inspect.isfunction(fn) or inspect.ismethod(fn) or inspect.isbuiltin(fn):
+        if (inspect.isfunction(fn) or inspect.ismethod(fn)
+                or inspect.isbuiltin(fn) or inspect.ismethodwrapper(fn)):
+            return False
+        fn = getattr(fn, "__call__", None)
+        if fn is None:
             return False
-        call = getattr(fn, "__call__", None)
-        return call is not None and inspect.iscoroutinefunction(inspect.unwrap(call))
 
 
 def _get_legacy_logging_warning_stacklevel() -> int:

tests/durabletask/test_coroutine_orchestrators.py

@@ -7,15 +7,17 @@
 import inspect
 import json
 import logging
+import subprocess
+import sys
 from collections.abc import AsyncGenerator, Generator
 from datetime import datetime, timedelta
-from functools import partial, wraps
+from functools import partial, update_wrapper, wraps
 from typing import Any
 
 import pytest
 
 from durabletask import entities, task, worker
-from durabletask.internal import helpers
+from durabletask.internal import helpers, shared
 from durabletask.internal import orchestrator_service_pb2 as pb
 from durabletask.serialization import JsonDataConverter
 
@@ -92,6 +94,75 @@ async def __call__(
     assert json.loads(complete.result.value) == 4
 
 
+@pytest.mark.parametrize("shape", [
+    "function", "partial", "bound-method", "callable", "partial-callable",
+    "async-callable", "nested-callable",
+])
+def test_async_orchestrator_is_recognized_before_its_synchronous_wrapped_function(shape: str) -> None:
+    def base(ctx: task.OrchestrationContext, value: int) -> int:
+        return value + 1
+
+    @wraps(base)
+    async def asynchronous(ctx: task.OrchestrationContext, value: int) -> int:
+        return base(ctx, value)
+
+    @wraps(asynchronous)
+    def traced(ctx: task.OrchestrationContext, value: int) -> Any:
+        return asynchronous(ctx, value)
+
+    class DecoratedCallable:
+        def base(self, ctx: task.OrchestrationContext, value: int) -> int:
+            return value + 1
+
+        async def asynchronous(self, ctx: task.OrchestrationContext, value: int) -> int:
+            return self.base(ctx, value)
+
+        def __call__(self, ctx: task.OrchestrationContext, value: int) -> Any:
+            return self.asynchronous(ctx, value)
+
+    update_wrapper(DecoratedCallable.asynchronous, DecoratedCallable.base)
+    update_wrapper(DecoratedCallable.__call__, DecoratedCallable.asynchronous)
+
+    class AsyncCallable(DecoratedCallable):
+        async def __call__(self, ctx: task.OrchestrationContext, value: int) -> int:
+            return self.base(ctx, value)
+
+    update_wrapper(AsyncCallable.__call__, AsyncCallable.base)
+
+    def wrap_callable(target: task.Orchestrator[Any, Any]) -> task.Orchestrator[Any, Any]:
+        class Wrapper:
+            def __call__(self, ctx: task.OrchestrationContext, value: int) -> Any:
+                return target(ctx, value)
+
+        update_wrapper(Wrapper.__call__, target)
+        return Wrapper()
+
+    decorated = DecoratedCallable()
+    nested: task.Orchestrator[Any, Any] = traced
+    # Each __call__ lookup produces a temporary bound method.
+    for _ in range(32):
+        nested = wrap_callable(nested)
+    variants = {
+        "function": traced,
+        "partial": partial(traced),
+        "bound-method": decorated.__call__,
+        "callable": decorated,
+        "partial-callable": partial(decorated),
+        "async-callable": AsyncCallable(),
+        "nested-callable": nested,
+    }
+    fn = variants[shape]
+    assert shared.is_async_callable(fn)
+    complete = _completion(_execute(fn, [], _start()))
+    assert complete.orchestrationStatus == pb.ORCHESTRATION_STATUS_COMPLETED
+    assert json.loads(complete.result.value) == 4
+
+
+@pytest.mark.parametrize("fn", [int, len, list[int].append, object().__str__])
+def test_synchronous_builtin_callables_are_not_classified_as_async(fn: Any) -> None:
+    assert not shared.is_async_callable(fn)
+
+
 def test_unmarked_synchronous_wrapper_returning_coroutine_is_rejected_and_closed() -> None:
     created: list[Any] = []
 
@@ -150,6 +221,47 @@ def wrapper(ctx: task.OrchestrationContext, value: int) -> int:
     assert "wrapper loop" in complete.failureDetails.errorMessage
 
 
+def _assert_mixed_wrapper_cycle_fails(shape: str) -> None:
+    def wrapper(ctx: task.OrchestrationContext, value: int) -> int:
+        return value
+
+    if shape == "partial":
+        wrapper.__wrapped__ = partial(wrapper)  # type: ignore[attr-defined]
+    else:
+        class CallableWrapper:
+            def __call__(self, ctx: task.OrchestrationContext, value: int) -> int:
+                return value
+
+        wrapper.__wrapped__ = CallableWrapper()  # type: ignore[attr-defined]
+        CallableWrapper.__call__.__wrapped__ = partial(wrapper)  # type: ignore[attr-defined]
+
+    # No input: dispatch must detect the cycle without relying on type discovery.
+    history = [
+        helpers.new_orchestrator_started_event(NOW),
+        helpers.new_execution_started_event("test", INSTANCE_ID),
+    ]
+    complete = _completion(_execute(wrapper, [], history))
+    assert complete.orchestrationStatus == pb.ORCHESTRATION_STATUS_FAILED
+    assert complete.failureDetails.errorType == "builtins.ValueError"
+    assert "wrapper loop" in complete.failureDetails.errorMessage
+    with pytest.raises(ValueError, match="wrapper loop"):
+        shared.is_async_callable(wrapper)
+
+
+@pytest.mark.parametrize("shape", ["partial", "callable-partial"])
+def test_mixed_wrapper_cycle_fails_within_bounded_worker_dispatch(shape: str) -> None:
+    result = subprocess.run(
+        [
+            sys.executable, "-c",
+            "import runpy, sys; "
+            "runpy.run_path(sys.argv[1])[sys.argv[2]](sys.argv[3])",
+            __file__, _assert_mixed_wrapper_cycle_fails.__name__, shape,
+        ],
+        capture_output=True, text=True, timeout=10,
+    )
+    assert result.returncode == 0, result.stdout + result.stderr
+
+
 def test_await_is_concrete_and_completed_tasks_do_not_suspend() -> None:
     assert "__await__" not in task.Task.__abstractmethods__