Address PR feedback (2)

Signed-off-by: Sergio Herrera <627709+seherv@users.noreply.github.com>

Author: seherv

ext/dapr-ext-workflow/AGENTS.md

@@ -116,7 +116,7 @@ At dispatch time (the gRPC stream loop in `_durabletask/worker.py`), `is_async_c
 
 Workflow (orchestrator) functions must remain generators (`def` with `yield`). They cannot be `async def` because durabletask's deterministic replay depends on synchronous generator semantics. Only activities support async.
 
-**Decorator ordering gotcha.** Stacking `@wfr.activity` over `@alternate_name(...)` over `async def` works because `@alternate_name` now emits an `async def innerfn` when the wrapped function is async. A user-written decorator that wraps an async function in a sync `def` (without `@functools.wraps` exposing `__wrapped__`) defeats `_is_async_callable`, routes the activity to the sync path, and produces an un-awaited coroutine. Such decorators should use `@functools.wraps(fn)` so the unwrap walks through them.
+**Decorator ordering gotcha.** Wrapping `@wfr.activity` over `@alternate_name(...)` over `async def` works because `@alternate_name` now emits an `async def innerfn` when the wrapped function is async. A user-written decorator that wraps an async function in a sync `def` (without `@functools.wraps` exposing `__wrapped__`) defeats `_is_async_callable`, routes the activity to the sync path, and produces an un-awaited coroutine. Such decorators should use `@functools.wraps(fn)` so the unwrap walks through them.
 
 **`maximum_thread_pool_workers` covers both paths.** This knob sizes the worker thread pool used for sync-activity bodies and for async-activity gRPC response sends. Mixed workloads with long-running sync activities can starve async response delivery (and vice versa) since they share the pool — size to the sum of peak sync activity concurrency and peak in-flight async response sends.
 

ext/dapr-ext-workflow/dapr/ext/workflow/_durabletask/internal/shared.py

@@ -21,6 +21,8 @@
 import grpc
 from dapr.ext.workflow import _model_protocol
 
+logger = logging.getLogger(__name__)
+
 
 def is_async_callable(fn: Any) -> bool:
     """Return True if ``fn`` is async. Catches ``functools.partial`` of coroutines,
@@ -35,7 +37,9 @@ def is_async_callable(fn: Any) -> bool:
         except ValueError:
             # Cyclic ``__wrapped__`` chain from a malformed decorator. Fall back to the
             # outermost callable; misclassification is preferable to crashing dispatch.
-            pass
+            logger.warning(
+                f'Cyclic __wrapped__ on {fn!r}, using outermost callable for async detection.'
+            )
     if inspect.iscoroutinefunction(candidate):
         return True
     if not inspect.isfunction(candidate) and hasattr(candidate, '__call__'):

ext/dapr-ext-workflow/dapr/ext/workflow/workflow_runtime.py

@@ -78,7 +78,11 @@ def _log_failure(ctx: task.ActivityContext, exc: Exception) -> None:
         activity_id = getattr(ctx, 'task_id', 'unknown')
         logger.warning(f'Activity execution failed - task_id: {activity_id}, error: {exc}')
 
-    if _is_async_callable(fn):
+    is_async = _is_async_callable(fn)
+    activity_name = getattr(fn, '__name__', repr(fn))
+    kind = 'async' if is_async else 'sync'
+    logger.debug(f"Registering activity '{activity_name}' on the {kind} dispatch path.")
+    if is_async:
 
         async def async_activity_wrapper(
             ctx: task.ActivityContext, inp: object | None = None

ext/dapr-ext-workflow/tests/durabletask/test_async_dispatch_regression.py

@@ -24,7 +24,7 @@
 
 pytestmark = pytest.mark.perf
 
-ACTIVITY_S = 0.02
+ACTIVITY_DURATION_SECONDS = 0.02
 N_ITEMS = 1000
 SEMAPHORE_CAP = 2000
 THREAD_POOL = 16
@@ -68,9 +68,10 @@ async def _run_async_batch(n_items: int, timeout_s: float) -> int:
     stub = _MockSidecarStub()
 
     async def activity(ctx, _inp) -> None:
-        await asyncio.sleep(ACTIVITY_S)
+        await asyncio.sleep(ACTIVITY_DURATION_SECONDS)
 
     worker_task = asyncio.create_task(manager.run())
+    # Non-blocking poll: yield to the event loop until the worker creates the activity queue
     while manager.activity_queue is None:
         await asyncio.sleep(0)
     try:
@@ -96,7 +97,7 @@ def test_async_activities_overlap_instead_of_serializing():
     try:
         completions = asyncio.run(_run_async_batch(N_ITEMS, TIMEOUT_S))
     except asyncio.TimeoutError:
-        serial_s = N_ITEMS * ACTIVITY_S
+        serial_s = N_ITEMS * ACTIVITY_DURATION_SECONDS
         pytest.fail(
             f'{N_ITEMS} async activities did not drain within {TIMEOUT_S:.1f}s. Serialized'
             f' they would cost ~{serial_s:.0f}s, so the async path is not overlapping I/O.'