Allow breakpoint() and input() through the sandbox in debug_mode

The workflow sandbox restricts the `breakpoint` and `input` builtins
as non-deterministic. With debug_mode set the user has explicitly opted
into single-stepping (and hence non-determinism for the duration of the
debug session), so calling breakpoint() in sandboxed workflow code
should reach the debugger rather than raising a
RestrictedWorkflowAccessError.

Adds a `_relax_sandbox_for_debugger` helper that, when debug_mode is
enabled and the runner is a SandboxedWorkflowRunner, rebuilds the
restrictions with `breakpoint` and `input` removed from the
`__builtins__` invalid-module-members set. Other restrictions are
untouched - this is a targeted relaxation, not full sandbox bypass.

Together with the existing dispatch fix (which moves activation onto
the main asyncio thread under debug_mode) this means a user can drop
`breakpoint()` into any workflow - sandboxed or not - and get an
interactive pdb prompt without swapping to UnsandboxedWorkflowRunner.

Adds test_breakpoint_works_in_sandboxed_workflow_in_debug_mode which
exercises the full path: sandboxed workflow + breakpoint() + worker
hook + main-thread placement. The test substitutes the captured
_ORIGINAL_BREAKPOINTHOOK with a stub so CI doesn't try to drive a real
pdb REPL.

4/4 tests pass on Python 3.13 and 3.14 locally. No regressions in
tests/worker/test_worker.py.

Author: elidlocke

temporalio/worker/_workflow.py

@@ -71,6 +71,43 @@ def _install_workflow_breakpoint_hook() -> None:
         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
@@ -119,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
@@ -150,11 +196,8 @@ def __init__(
         )
         self._throw_after_activation: Exception | None = None
 
-        # Debug mode also enabled by the TEMPORAL_DEBUG env var. In debug mode,
-        # deadlock detection is disabled and workflow activations run inline on
-        # the asyncio main thread so interactive debuggers (pdb, breakpoint(),
-        # IDE debuggers) can read stdin.
-        self._debug_mode = bool(debug_mode or os.environ.get("TEMPORAL_DEBUG"))
+        # 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

tests/worker/test_breakpoint_hang.py

@@ -17,10 +17,12 @@
 import sys
 import threading
 import uuid
+from unittest.mock import patch
 
 from temporalio import workflow
 from temporalio.client import Client
 from temporalio.worker import Worker
+from temporalio.worker import _workflow as _workflow_module
 from temporalio.worker._workflow import _temporal_workflow_breakpoint_hook
 
 
@@ -87,6 +89,61 @@ async def test_workflow_runs_on_main_thread_in_debug_mode(client: Client):
     )
 
 
+@workflow.defn
+class SandboxedBreakpointWorkflow:
+    """Sandboxed workflow that calls breakpoint() — verifies the fix works
+    without requiring users to switch to UnsandboxedWorkflowRunner."""
+
+    @workflow.run
+    async def run(self) -> str:
+        breakpoint()
+        return "done"
+
+
+async def test_breakpoint_works_in_sandboxed_workflow_in_debug_mode(client: Client):
+    """`breakpoint()` inside a *sandboxed* workflow should reach the debugger
+    when `debug_mode=True` is set. This is the natural unit-test scenario:
+    users want to drop `breakpoint()` into their workflow without having to
+    swap the runner.
+
+    We can't drive a real pdb REPL in CI, so we substitute the captured
+    `_ORIGINAL_BREAKPOINTHOOK` (which our worker hook delegates to after the
+    thread check) with a stub that records the call. If the stub is reached,
+    we've proven the path from `breakpoint()` → our hook → original hook
+    works through the sandbox, on the main thread.
+    """
+    captured: dict[str, object] = {}
+
+    def stub_breakpointhook(*args: object, **kwargs: object) -> None:
+        captured["thread"] = threading.current_thread().name
+        captured["called"] = True
+
+    task_queue = f"tq-{uuid.uuid4()}"
+    with patch.object(
+        _workflow_module, "_ORIGINAL_BREAKPOINTHOOK", stub_breakpointhook
+    ):
+        async with Worker(
+            client,
+            task_queue=task_queue,
+            workflows=[SandboxedBreakpointWorkflow],
+            debug_mode=True,
+        ):
+            result = await client.execute_workflow(
+                SandboxedBreakpointWorkflow.run,
+                id=f"wf-{uuid.uuid4()}",
+                task_queue=task_queue,
+            )
+
+    assert result == "done", (
+        f"workflow did not complete; breakpoint() likely raised inside the sandbox: "
+        f"result={result!r}"
+    )
+    assert captured.get("called"), "breakpoint hook was never reached"
+    assert captured["thread"] == threading.main_thread().name, (
+        f"breakpoint hook ran on {captured['thread']!r}, not the main thread"
+    )
+
+
 def test_breakpoint_hook_raises_on_workflow_thread(client: Client):
     """The defensive hook fails loudly when breakpoint() is called from a
     `temporal_workflow_*` thread without debug mode."""