Make workflow deadlock detection timeout configurable

The workflow-task deadlock detector timeout was hardcoded to 2 seconds,
with `debug_mode` / `TEMPORAL_DEBUG` as the only escape hatch — and that
disables detection entirely (and may relax the sandbox). There was no way
to keep detection on but grant more headroom to workflows that legitimately
do heavy CPU-bound work on the workflow thread before yielding.

Add a `deadlock_detection_timeout: timedelta = timedelta(seconds=2)` option
to `Worker` and `Replayer`, plumbed through to `_WorkflowWorker`. The default
preserves the current 2s behavior; `debug_mode` still disables detection.
This matches the Go SDK's `WorkerOptions.DeadlockDetectionTimeout`.

Author: William Ma

CHANGELOG.md

@@ -20,6 +20,13 @@ to include examples, links to docs, or any other relevant information.
 
 ### Added
 
+- `Worker` and `Replayer` now accept a `deadlock_detection_timeout`
+  (`timedelta`, default 2 seconds) to configure how long a workflow task may run
+  without yielding before it is failed as a deadlock (`[TMPRL1101]`). This brings
+  the Python SDK in line with the Go SDK's `WorkerOptions.DeadlockDetectionTimeout`.
+  Raise it only for workflows that legitimately perform heavy CPU-bound work on
+  the workflow thread; `debug_mode` still disables detection entirely.
+
 ### Changed
 
 - AWS Lambda worker `configure` parameter supports sync, async, and async

temporalio/worker/_replayer.py

@@ -8,6 +8,7 @@
 from collections.abc import AsyncIterator, Mapping, Sequence
 from contextlib import AbstractAsyncContextManager, asynccontextmanager
 from dataclasses import dataclass
+from datetime import timedelta
 
 from typing_extensions import TypedDict
 
@@ -48,6 +49,7 @@ def __init__(
         identity: str | None = None,
         workflow_failure_exception_types: Sequence[type[BaseException]] = [],
         debug_mode: bool = False,
+        deadlock_detection_timeout: timedelta = timedelta(seconds=2),
         runtime: temporalio.runtime.Runtime | None = None,
         disable_safe_workflow_eviction: bool = False,
         header_codec_behavior: HeaderCodecBehavior = HeaderCodecBehavior.NO_CODEC,
@@ -78,6 +80,7 @@ def __init__(
             identity=identity,
             workflow_failure_exception_types=workflow_failure_exception_types,
             debug_mode=debug_mode,
+            deadlock_detection_timeout=deadlock_detection_timeout,
             runtime=runtime,
             disable_safe_workflow_eviction=disable_safe_workflow_eviction,
             header_codec_behavior=header_codec_behavior,
@@ -256,6 +259,9 @@ def on_eviction_hook(
                     "workflow_failure_exception_types", []
                 ),
                 debug_mode=self._config.get("debug_mode", False),
+                deadlock_detection_timeout=self._config.get(
+                    "deadlock_detection_timeout", timedelta(seconds=2)
+                ),
                 metric_meter=runtime.metric_meter,
                 on_eviction_hook=on_eviction_hook,
                 disable_eager_activity_execution=False,
@@ -410,6 +416,7 @@ class ReplayerConfig(TypedDict, total=False):
     identity: str | None
     workflow_failure_exception_types: Sequence[type[BaseException]]
     debug_mode: bool
+    deadlock_detection_timeout: timedelta
     runtime: temporalio.runtime.Runtime | None
     disable_safe_workflow_eviction: bool
     header_codec_behavior: HeaderCodecBehavior

temporalio/worker/_worker.py

@@ -130,6 +130,7 @@ def __init__(
         workflow_failure_exception_types: Sequence[type[BaseException]] = [],
         shared_state_manager: SharedStateManager | None = None,
         debug_mode: bool = False,
+        deadlock_detection_timeout: timedelta = timedelta(seconds=2),
         disable_eager_activity_execution: bool = False,
         on_fatal_error: Callable[[BaseException], Awaitable[None]] | None = None,
         use_worker_versioning: bool = False,
@@ -282,6 +283,14 @@ def __init__(
                 sandboxing in order to make using a debugger easier. If false
                 but the environment variable ``TEMPORAL_DEBUG`` is truthy, this
                 will be set to true.
+            deadlock_detection_timeout: Maximum amount of time a workflow task
+                is allowed to run without yielding control back to the event
+                loop before it is considered a deadlock and the task fails with
+                ``[TMPRL1101]``. Defaults to 2 seconds. Increase this only for
+                workflows that legitimately perform heavy CPU-bound work on the
+                workflow thread; raising it weakens the deadlock detector's
+                ability to catch genuinely stuck workflows. Ignored when
+                ``debug_mode`` is enabled (detection is fully disabled then).
             disable_eager_activity_execution: If true, will disable eager
                 activity execution. Eager activity execution is an optimization
                 on some servers that sends activities back to the same worker as
@@ -363,6 +372,7 @@ def __init__(
             workflow_failure_exception_types=workflow_failure_exception_types,
             shared_state_manager=shared_state_manager,
             debug_mode=debug_mode,
+            deadlock_detection_timeout=deadlock_detection_timeout,
             disable_eager_activity_execution=disable_eager_activity_execution,
             on_fatal_error=on_fatal_error,
             use_worker_versioning=use_worker_versioning,
@@ -529,6 +539,7 @@ def check_activity(activity: str):
                     "workflow_failure_exception_types"
                 ],  # type: ignore[reportTypedDictNotRequiredAccess]
                 debug_mode=config["debug_mode"],  # type: ignore[reportTypedDictNotRequiredAccess]
+                deadlock_detection_timeout=config["deadlock_detection_timeout"],  # type: ignore[reportTypedDictNotRequiredAccess]
                 disable_eager_activity_execution=config[
                     "disable_eager_activity_execution"
                 ],  # type: ignore[reportTypedDictNotRequiredAccess]
@@ -977,6 +988,7 @@ class WorkerConfig(TypedDict, total=False):
     workflow_failure_exception_types: Sequence[type[BaseException]]
     shared_state_manager: SharedStateManager | None
     debug_mode: bool
+    deadlock_detection_timeout: timedelta
     disable_eager_activity_execution: bool
     on_fatal_error: Callable[[BaseException], Awaitable[None]] | None
     use_worker_versioning: bool

temporalio/worker/_workflow.py

@@ -79,6 +79,7 @@ def __init__(
         interceptors: Sequence[Interceptor],
         workflow_failure_exception_types: Sequence[type[BaseException]],
         debug_mode: bool,
+        deadlock_detection_timeout: timedelta,
         disable_eager_activity_execution: bool,
         metric_meter: temporalio.common.MetricMeter,
         on_eviction_hook: Callable[
@@ -155,9 +156,11 @@ def __init__(
         )
         self._throw_after_activation: Exception | None = None
 
-        # If debug mode is enabled, disable deadlock detection
-        # otherwise set to 2 seconds
-        self._deadlock_timeout_seconds = None if self._debug_mode else 2
+        # If debug mode is enabled, disable deadlock detection entirely;
+        # otherwise use the configured timeout (defaults to 2 seconds).
+        self._deadlock_timeout_seconds = (
+            None if self._debug_mode else deadlock_detection_timeout.total_seconds()
+        )
 
         # Keep track of workflows that could not be evicted
         self._could_not_evict_count = 0

tests/worker/test_worker.py

@@ -1707,3 +1707,35 @@ def debug_envvar():
             "_pydevd_bundle"
             in worker._workflow_worker._workflow_runner.restrictions.passthrough_modules
         )
+
+
+async def test_worker_deadlock_detection_timeout(client: Client):
+    # Default is 2 seconds.
+    worker = Worker(
+        client,
+        workflows=[SimpleWorkflow],
+        task_queue=f"task-queue-{uuid.uuid4()}",
+    )
+    assert worker._workflow_worker
+    assert worker._workflow_worker._deadlock_timeout_seconds == 2
+
+    # A custom timeout is honored.
+    worker = Worker(
+        client,
+        workflows=[SimpleWorkflow],
+        task_queue=f"task-queue-{uuid.uuid4()}",
+        deadlock_detection_timeout=timedelta(seconds=10),
+    )
+    assert worker._workflow_worker
+    assert worker._workflow_worker._deadlock_timeout_seconds == 10
+
+    # debug_mode disables detection regardless of the configured timeout.
+    worker = Worker(
+        client,
+        workflows=[SimpleWorkflow],
+        task_queue=f"task-queue-{uuid.uuid4()}",
+        deadlock_detection_timeout=timedelta(seconds=10),
+        debug_mode=True,
+    )
+    assert worker._workflow_worker
+    assert worker._workflow_worker._deadlock_timeout_seconds is None