Fix cancellations being swallowed in some circumstances

CHANGELOG.md

@@ -33,6 +33,15 @@ to include examples, links to docs, or any other relevant information.
 
 ### Changed
 
+- Prepared replay-safe workflow activation scheduling that prevents cancellation
+  from being lost when another event becomes ready in the same workflow task. The
+  behavior is guarded by internal workflow logic flag 2 and remains disabled by
+  default during its compatibility rollout.
+  **Maintainer reminder:** keep flag 2 default-disabled for the first two published
+  SDK releases that recognize it; enable it in the third release, remove the explicit
+  overrides for this flag from `tests/worker/test_workflow.py`, and replace this rollout
+  note with a `Fixed` entry announcing the behavior change.
+
 ### Deprecated
 
 ### Breaking Changes

temporalio/worker/_replayer.py

@@ -24,7 +24,12 @@
 from ._interceptor import Interceptor
 from ._worker import load_default_build_id
 from ._workflow import _WorkflowWorker
-from ._workflow_instance import UnsandboxedWorkflowRunner, WorkflowRunner
+from ._workflow_instance import (
+    _DEFAULT_ENABLED_WORKFLOW_LOGIC_FLAGS,
+    UnsandboxedWorkflowRunner,
+    WorkflowRunner,
+    _WorkflowLogicFlag,
+)
 from .workflow_sandbox import SandboxedWorkflowRunner
 
 logger = logging.getLogger(__name__)
@@ -83,6 +88,7 @@ def __init__(
             header_codec_behavior=header_codec_behavior,
         )
         self._initial_config = self._config.copy()
+        self._default_workflow_logic_flags = set(_DEFAULT_ENABLED_WORKFLOW_LOGIC_FLAGS)
 
         # Apply plugin configuration
         self.plugins = plugins
@@ -93,6 +99,14 @@ def __init__(
         if not self._config.get("workflows"):
             raise ValueError("At least one workflow must be specified")
 
+    def _set_default_workflow_logic_flag(
+        self, flag: _WorkflowLogicFlag, *, enabled: bool
+    ) -> None:
+        if enabled:
+            self._default_workflow_logic_flags.add(flag)
+        else:
+            self._default_workflow_logic_flags.discard(flag)
+
     def config(self, *, active_config: bool = False) -> ReplayerConfig:
         """Config, as a dictionary, used to create this replayer.
 
@@ -270,6 +284,9 @@ def on_eviction_hook(
                 )
                 != HeaderCodecBehavior.NO_CODEC,
                 max_workflow_task_external_storage_concurrency=1,
+                default_workflow_logic_flags=frozenset(
+                    self._default_workflow_logic_flags
+                ),
             )
             external_storage = data_converter.external_storage
             storage_driver_types = (

temporalio/worker/_worker.py

@@ -43,6 +43,7 @@
     PatchActivationInput,
     UnsandboxedWorkflowRunner,
     WorkflowRunner,
+    _WorkflowLogicFlag,
 )
 from .workflow_sandbox import SandboxedWorkflowRunner
 
@@ -735,6 +736,17 @@ def client(self, value: temporalio.client.Client) -> None:
         if self._nexus_worker:
             self._nexus_worker._client = value
 
+    def _set_default_workflow_logic_flag(
+        self, flag: _WorkflowLogicFlag, *, enabled: bool
+    ) -> None:
+        if self._started:
+            raise RuntimeError(
+                "Cannot set default workflow logic flags after the worker has started"
+            )
+        if not self._workflow_worker:
+            raise RuntimeError("Cannot set workflow logic flags without workflows")
+        self._workflow_worker._set_default_workflow_logic_flag(flag, enabled=enabled)
+
     @property
     def is_running(self) -> bool:
         """Whether the worker is running.

temporalio/worker/_workflow.py

@@ -40,11 +40,13 @@
     WorkflowInterceptorClassInput,
 )
 from ._workflow_instance import (
+    _DEFAULT_ENABLED_WORKFLOW_LOGIC_FLAGS,
     PatchActivationInput,
     WorkflowInstance,
     WorkflowInstanceDetails,
     WorkflowRunner,
     _WorkflowExternFunctions,
+    _WorkflowLogicFlag,
 )
 
 logger = logging.getLogger(__name__)
@@ -90,13 +92,19 @@ def __init__(
         assert_local_activity_valid: Callable[[str], None],
         encode_headers: bool,
         max_workflow_task_external_storage_concurrency: int,
+        default_workflow_logic_flags: frozenset[_WorkflowLogicFlag] | None = None,
     ) -> None:
         # Debug mode is enabled if specified or if the TEMPORAL_DEBUG env var is truthy
         debug_mode = debug_mode or bool(os.environ.get("TEMPORAL_DEBUG"))
 
         self._bridge_worker = bridge_worker
         self._namespace = namespace
         self._task_queue = task_queue
+        self._default_workflow_logic_flags = set(
+            _DEFAULT_ENABLED_WORKFLOW_LOGIC_FLAGS
+            if default_workflow_logic_flags is None
+            else default_workflow_logic_flags
+        )
         self._workflow_task_executor = (
             workflow_task_executor
             or concurrent.futures.ThreadPoolExecutor(
@@ -788,6 +796,7 @@ def _create_workflow_instance(
             patch_activation_callback=self._patch_activation_callback,
             last_completion_result=init.last_completion_result,
             last_failure=last_failure,
+            default_workflow_logic_flags=frozenset(self._default_workflow_logic_flags),
         )
         if defn.sandboxed:
             return self._workflow_runner.create_instance(det)
@@ -800,6 +809,14 @@ def nondeterminism_as_workflow_fail(self) -> bool:
             for typ in self._workflow_failure_exception_types
         )
 
+    def _set_default_workflow_logic_flag(
+        self, flag: _WorkflowLogicFlag, *, enabled: bool
+    ) -> None:
+        if enabled:
+            self._default_workflow_logic_flags.add(flag)
+        else:
+            self._default_workflow_logic_flags.discard(flag)
+
     def nondeterminism_as_workflow_fail_for_types(self) -> set[str]:
         return {
             k