Fix pdb / breakpoint() hang in workflow code (temporalio#1104)

Closes temporalio#1104.

breakpoint() and pdb.set_trace() inside workflow code silently hang
even with debug_mode=True and an unsandboxed runner. Three orthogonal
issues contribute; this PR addresses all three behind the existing
debug_mode flag so production behavior is unchanged.

1. Thread placement. Activations run on a ThreadPoolExecutor worker
   thread, so pdb's cmdloop() calls input() from a thread that doesn't
   own the controlling TTY. Fixed by scheduling the activation as a
   loop.call_soon callback and awaiting a future the callback
   completes. The dispatch task suspends at the await so it's no
   longer mid-__step() when the workflow's internal task stepping
   happens. (A direct synchronous call ran afoul of Python 3.14's
   tightened asyncio task-entry validation: "Cannot enter into task
   while another task is being executed.")

2. Sandbox restriction. The sandbox flags `breakpoint` and `input` as
   non-deterministic builtins. With debug_mode=True the user has
   explicitly accepted non-determinism for the debugging session, so
   we relax those two specific restrictions when the runner is a
   SandboxedWorkflowRunner. Other sandbox checks remain intact.

3. Silent-hang failure mode. Installs a process-wide
   sys.breakpointhook at worker startup that raises a clear
   RuntimeError when breakpoint() is called from a workflow worker
   thread without debug_mode, replacing the silent hang.

Adds a "Debugging Workflows with breakpoint() / pdb" subsection to the
README under Workflow Sandbox, including a runnable example and the
caveats around workflow task timeouts.

Tests at tests/worker/test_breakpoint_hang.py cover thread placement
in both modes, the sandboxed-workflow path, and the defensive hook.
Verified on Python 3.13 and 3.14 locally; CI matrix green on fork.

The load-bearing observation for the dispatch fix: `await future`
suspends the dispatch task such that asyncio no longer considers it
"currently executing," even though it's still in a pending state.
That's what lets workflow.activate(act) step the workflow's internal
task without 3.14's task-entry error.

Author: elidlocke

README.md

@@ -82,6 +82,7 @@ informal introduction to the features and their implementation.
         - [Customizing the Sandbox](#customizing-the-sandbox)
           - [Passthrough Modules](#passthrough-modules)
           - [Invalid Module Members](#invalid-module-members)
+        - [Debugging Workflows with `breakpoint()` / `pdb`](#debugging-workflows-with-breakpoint--pdb)
         - [Known Sandbox Issues](#known-sandbox-issues)
           - [Global Import/Builtins](#global-importbuiltins)
           - [Sandbox is not Secure](#sandbox-is-not-secure)
@@ -1241,6 +1242,75 @@ my_worker = Worker(..., workflow_runner=SandboxedWorkflowRunner(restrictions=my_
 
 See the API for more details on exact fields and their meaning.
 
+##### Debugging Workflows with `breakpoint()` / `pdb`
+
+Setting `debug_mode=True` on the `Worker` (or `TEMPORAL_DEBUG=1` in the environment) routes workflow activations
+onto the asyncio main thread instead of a worker thread pool. This lets `breakpoint()` and `pdb.set_trace()`
+inside workflow code open an interactive REPL — without it, pdb hangs because its `input()` call would run on a
+thread that does not own the controlling TTY.
+
+A minimal runnable example:
+
+```python
+import asyncio
+from datetime import timedelta
+
+from temporalio import workflow
+from temporalio.client import Client
+from temporalio.worker import Worker
+
+
+@workflow.defn(sandboxed=False)
+class DebugMeWorkflow:
+    @workflow.run
+    async def run(self) -> str:
+        x = 42
+        breakpoint()  # interactive pdb prompt opens here
+        return f"x was {x}"
+
+
+async def main() -> None:
+    client = await Client.connect("localhost:7233")
+    async with Worker(
+        client,
+        task_queue="debug-me",
+        workflows=[DebugMeWorkflow],
+        debug_mode=True,
+    ):
+        result = await client.execute_workflow(
+            DebugMeWorkflow.run,
+            id="debug-me-wf",
+            task_queue="debug-me",
+            task_timeout=timedelta(minutes=10),  # see caveat below
+        )
+        print(result)
+
+
+if __name__ == "__main__":
+    asyncio.run(main())
+```
+
+Run with `python debug_me.py` (not under pytest, which captures stdin and breaks the REPL). At the `(Pdb)`
+prompt try `p x`, `n`, `c`, `q`.
+
+Two caveats when pausing at a breakpoint inside a workflow:
+
+1. **Workflow task timeout.** Temporal expires a workflow task after ~10 seconds by default. If you sit at the
+   `(Pdb)` prompt longer than that, the server reassigns the task and your workflow replays from the start when
+   you continue — re-hitting the breakpoint. Pass `task_timeout=timedelta(minutes=N)` to `execute_workflow` /
+   `start_workflow` to give yourself debugging headroom:
+
+   ```python
+   await client.execute_workflow(MyWorkflow.run, ..., task_timeout=timedelta(minutes=10))
+   ```
+
+2. **Deterministic replay.** Workflows are deterministic and replay from history; any wall-clock pause violates
+   that contract. For post-mortem debugging without these caveats, use the [Replayer](#replayer) on a recorded
+   history instead of live debugging.
+
+A `breakpoint()` call from workflow code without `debug_mode` enabled raises a `RuntimeError` with a pointer to
+this section, so the failure mode is loud rather than a silent hang.
+
 ##### Known Sandbox Issues
 
 Below are known sandbox issues. As the sandbox is developed and matures, some may be resolved.

temporalio/worker/_workflow.py

@@ -48,6 +48,66 @@
 # Set to true to log all activations and completions
 LOG_PROTOS = False
 
+# Name prefix for threads in the workflow task ThreadPoolExecutor. Used by the
+# breakpoint hook below to detect calls coming from workflow code.
+_WORKFLOW_THREAD_NAME_PREFIX = "temporal_workflow_"
+
+_ORIGINAL_BREAKPOINTHOOK = sys.breakpointhook
+
+
+def _temporal_workflow_breakpoint_hook(*args: object, **kwargs: object) -> object:
+    if threading.current_thread().name.startswith(_WORKFLOW_THREAD_NAME_PREFIX):
+        raise RuntimeError(
+            "breakpoint() / pdb.set_trace() inside workflow code requires "
+            "debug_mode=True (or the TEMPORAL_DEBUG environment variable) on "
+            "the Worker. Without it the workflow runs on a thread pool and "
+            "pdb's interactive REPL cannot read stdin."
+        )
+    return _ORIGINAL_BREAKPOINTHOOK(*args, **kwargs)
+
+
+def _install_workflow_breakpoint_hook() -> None:
+    if sys.breakpointhook is not _temporal_workflow_breakpoint_hook:
+        sys.breakpointhook = _temporal_workflow_breakpoint_hook
+
+
+def _relax_sandbox_for_debugger(workflow_runner: WorkflowRunner) -> WorkflowRunner:
+    """Remove the sandbox restrictions on `breakpoint` and `input` so pdb /
+    breakpoint() can be used inside sandboxed workflow code under debug_mode.
+
+    The sandbox flags both as non-deterministic builtins by default. Users
+    who opt into debug_mode have explicitly accepted non-determinism for a
+    debugging session, so we relax these specific restrictions instead of
+    forcing users to switch to `UnsandboxedWorkflowRunner` (which also
+    disables every other sandbox check, a larger blast radius).
+
+    Returns the runner unchanged if it isn't a SandboxedWorkflowRunner.
+    """
+    # Import lazily so users without the sandbox module aren't penalized.
+    from temporalio.worker.workflow_sandbox._runner import SandboxedWorkflowRunner
+
+    if not isinstance(workflow_runner, SandboxedWorkflowRunner):
+        return workflow_runner
+
+    restrictions = workflow_runner.restrictions
+    invalid = restrictions.invalid_module_members
+    builtins_matcher = invalid.children.get("__builtins__")
+    if builtins_matcher is None or not (
+        "breakpoint" in builtins_matcher.use or "input" in builtins_matcher.use
+    ):
+        return workflow_runner
+
+    new_use = set(builtins_matcher.use) - {"breakpoint", "input"}
+    new_builtins = dataclasses.replace(builtins_matcher, use=new_use)
+    new_invalid = dataclasses.replace(
+        invalid, children={**invalid.children, "__builtins__": new_builtins}
+    )
+    new_restrictions = dataclasses.replace(
+        restrictions, invalid_module_members=new_invalid
+    )
+    return dataclasses.replace(workflow_runner, restrictions=new_restrictions)
+
+
 # Value was chosen abitrarily as a small number that allows some concurrency and prevents
 # large numbers of concurrent external storage operations causing resource contention.
 # This default limit is per workflow task activation and does not limit the total number
@@ -96,6 +156,15 @@ def __init__(
             )
         )
         self._workflow_task_executor_user_provided = workflow_task_executor is not None
+
+        # Debug mode also enabled by the TEMPORAL_DEBUG env var. In debug mode,
+        # deadlock detection is disabled, workflow activations run inline on
+        # the asyncio main thread so interactive debuggers (pdb, breakpoint(),
+        # IDE debuggers) can read stdin, and the sandbox restriction on
+        # `breakpoint`/`input` is lifted so calls reach the debugger.
+        self._debug_mode = bool(debug_mode or os.environ.get("TEMPORAL_DEBUG"))
+        if self._debug_mode:
+            workflow_runner = _relax_sandbox_for_debugger(workflow_runner)
         self._workflow_runner = workflow_runner
         self._unsandboxed_workflow_runner = unsandboxed_workflow_runner
         self._data_converter = data_converter
@@ -127,11 +196,15 @@ def __init__(
         )
         self._throw_after_activation: Exception | None = None
 
-        # If there's a debug mode or a truthy TEMPORAL_DEBUG env var, disable
-        # deadlock detection, otherwise set to 2 seconds
-        self._deadlock_timeout_seconds = (
-            None if debug_mode or os.environ.get("TEMPORAL_DEBUG") else 2
-        )
+        # self._debug_mode is set earlier (before workflow_runner is assigned)
+        # so the sandbox relaxation can take effect on the runner.
+        self._deadlock_timeout_seconds = None if self._debug_mode else 2
+
+        # Install a process-wide breakpoint hook that fails loudly when
+        # breakpoint() is called from a workflow worker thread without debug
+        # mode. Without this guard the call silently hangs because pdb's
+        # input() runs off the main thread.
+        _install_workflow_breakpoint_hook()
 
         # Keep track of workflows that could not be evicted
         self._could_not_evict_count = 0
@@ -241,6 +314,37 @@ async def drain_poll_queue(self) -> None:
             except PollShutdownError:
                 return
 
+    async def _activate_inline_for_debug(
+        self,
+        loop: asyncio.AbstractEventLoop,
+        workflow: _RunningWorkflow,
+        act: temporalio.bridge.proto.workflow_activation.WorkflowActivation,
+    ) -> temporalio.bridge.proto.workflow_completion.WorkflowActivationCompletion:
+        # Run `workflow.activate(act)` synchronously, but indirected through
+        # `loop.call_soon` so it executes outside the dispatch task's __step()
+        # context. Python 3.14's asyncio refuses to enter a task while another
+        # task on the same thread is mid-step; awaiting the future below
+        # suspends this dispatch task so the workflow's own task stepping does
+        # not collide with it.
+        future: asyncio.Future = loop.create_future()
+
+        def run_inline() -> None:
+            # The workflow's _run_once clears the running-loop registration to
+            # None when it exits. Save the main loop here and restore it after
+            # the activation so any code that runs subsequently still sees the
+            # right loop.
+            main_loop = asyncio._get_running_loop()
+            try:
+                completion = workflow.activate(act)
+                future.set_result(completion)
+            except BaseException as e:
+                future.set_exception(e)
+            finally:
+                asyncio._set_running_loop(main_loop)
+
+        loop.call_soon(run_inline)
+        return await future
+
     async def _handle_activation(
         self, act: temporalio.bridge.proto.workflow_activation.WorkflowActivation
     ) -> None:
@@ -330,35 +434,52 @@ async def _handle_activation(
                 )
                 self._running_workflows[act.run_id] = workflow
 
-            # Run activation in separate thread so we can check if it's
-            # deadlocked
-            activate_task = asyncio.get_running_loop().run_in_executor(
-                self._workflow_task_executor,
-                workflow.activate,
-                act,
-            )
-
-            # Run activation task with deadlock timeout
-            try:
-                completion = await asyncio.wait_for(
-                    activate_task, self._deadlock_timeout_seconds
+            if self._debug_mode:
+                # Run activation inline on the asyncio main thread so
+                # interactive debuggers (pdb, breakpoint(), IDE debuggers)
+                # can read stdin. We schedule the synchronous activation as
+                # a `call_soon` callback rather than calling it directly,
+                # so the dispatch task suspends at the `await` below and is
+                # no longer mid-`__step()` when the workflow's own task
+                # gets stepped inside `workflow.activate`. Python 3.14's
+                # asyncio refuses to enter a task while another task on
+                # the same thread is currently being executed; the
+                # call_soon detour avoids that nesting. The main loop is
+                # still blocked while the activation runs, which is the
+                # intended behavior when debugging.
+                completion = await self._activate_inline_for_debug(
+                    asyncio.get_running_loop(), workflow, act
                 )
-            except asyncio.TimeoutError:
-                # Need to create the deadlock exception up here so it
-                # captures the trace now instead of later after we may have
-                # interrupted it
-                deadlock_exc = _DeadlockError.from_deadlocked_workflow(
-                    workflow.instance, self._deadlock_timeout_seconds
+            else:
+                # Run activation in separate thread so we can check if it's
+                # deadlocked
+                activate_task = asyncio.get_running_loop().run_in_executor(
+                    self._workflow_task_executor,
+                    workflow.activate,
+                    act,
                 )
-                # When we deadlock, we will raise an exception to fail
-                # the task. But before we do that, we want to try to
-                # interrupt the thread and put this activation task on
-                # the workflow so that the successive eviction can wait
-                # on it before trying to evict.
-                workflow.attempt_deadlock_interruption()
-                # Set the task and raise
-                workflow.deadlocked_activation_task = activate_task
-                raise deadlock_exc from None
+
+                # Run activation task with deadlock timeout
+                try:
+                    completion = await asyncio.wait_for(
+                        activate_task, self._deadlock_timeout_seconds
+                    )
+                except asyncio.TimeoutError:
+                    # Need to create the deadlock exception up here so it
+                    # captures the trace now instead of later after we may have
+                    # interrupted it
+                    deadlock_exc = _DeadlockError.from_deadlocked_workflow(
+                        workflow.instance, self._deadlock_timeout_seconds
+                    )
+                    # When we deadlock, we will raise an exception to fail
+                    # the task. But before we do that, we want to try to
+                    # interrupt the thread and put this activation task on
+                    # the workflow so that the successive eviction can wait
+                    # on it before trying to evict.
+                    workflow.attempt_deadlock_interruption()
+                    # Set the task and raise
+                    workflow.deadlocked_activation_task = activate_task
+                    raise deadlock_exc from None
 
         except Exception as err:
             if isinstance(err, _DeadlockError):
@@ -576,22 +697,30 @@ async def _handle_cache_eviction(
             handle_eviction_task: asyncio.Future | None = None
             while True:
                 try:
-                    # We only create the eviction task if we haven't already or
-                    # it is done. This is because if it already is running and
-                    # timed out, it's still running (and holding on to a
-                    # thread). But if did complete running but failed with
-                    # another error, we want to re-create the task.
-                    if not handle_eviction_task or handle_eviction_task.done():
-                        handle_eviction_task = (
-                            asyncio.get_running_loop().run_in_executor(
-                                self._workflow_task_executor,
-                                workflow.activate,
-                                act,
+                    if self._debug_mode:
+                        # Eviction activation runs inline on the main thread
+                        # too. See `_activate_inline_for_debug` for why the
+                        # `call_soon` detour is needed on Python 3.14+.
+                        await self._activate_inline_for_debug(
+                            asyncio.get_running_loop(), workflow, act
+                        )
+                    else:
+                        # We only create the eviction task if we haven't already or
+                        # it is done. This is because if it already is running and
+                        # timed out, it's still running (and holding on to a
+                        # thread). But if did complete running but failed with
+                        # another error, we want to re-create the task.
+                        if not handle_eviction_task or handle_eviction_task.done():
+                            handle_eviction_task = (
+                                asyncio.get_running_loop().run_in_executor(
+                                    self._workflow_task_executor,
+                                    workflow.activate,
+                                    act,
+                                )
                             )
+                        await asyncio.wait_for(
+                            handle_eviction_task, self._deadlock_timeout_seconds
                         )
-                    await asyncio.wait_for(
-                        handle_eviction_task, self._deadlock_timeout_seconds
-                    )
                     # Break if it succeeds
                     break
                 except BaseException as err: