PR feedback, fix order of args, remove heartbeat=0 check

Author: yuandrew

temporalio/bridge/src/worker.rs

@@ -675,6 +675,7 @@ impl WorkerRef {
     }
 
     fn replace_client(&self, client: &client::ClientRef) -> PyResult<()> {
+        enter_sync!(self.runtime);
         self.worker
             .as_ref()
             .expect("missing worker")

temporalio/client.py

@@ -242,10 +242,10 @@ def __init__(
             service_client=service_client,
             namespace=namespace,
             data_converter=data_converter,
+            plugins=plugins,
             interceptors=interceptors,
             default_workflow_query_reject_condition=default_workflow_query_reject_condition,
             header_codec_behavior=header_codec_behavior,
-            plugins=plugins,
         )
 
         for plugin in plugins:
@@ -1560,12 +1560,12 @@ class ClientConfig(TypedDict, total=False):
     service_client: Required[temporalio.service.ServiceClient]
     namespace: Required[str]
     data_converter: Required[temporalio.converter.DataConverter]
+    plugins: Required[Sequence[Plugin]]
     interceptors: Required[Sequence[Interceptor]]
     default_workflow_query_reject_condition: Required[
         Optional[temporalio.common.QueryRejectCondition]
     ]
     header_codec_behavior: Required[HeaderCodecBehavior]
-    plugins: Required[Sequence[Plugin]]
 
 
 class WorkflowHistoryEventFilterType(IntEnum):

temporalio/runtime.py

@@ -131,7 +131,7 @@ def __init__(
             telemetry: Telemetry configuration when not supplying
                 ``runtime_options``.
             worker_heartbeat_interval: Interval for worker heartbeats. ``None``
-                disables heartbeating.
+                disables heartbeating. Interval must be between 1s and 60s.
 
         Raises:
             ValueError: If both ```runtime_options`` is a negative value.
@@ -142,10 +142,6 @@ def __init__(
             if worker_heartbeat_interval <= timedelta(0):
                 raise ValueError("worker_heartbeat_interval must be positive")
             heartbeat_millis = int(worker_heartbeat_interval.total_seconds() * 1000)
-            if heartbeat_millis == 0:
-                heartbeat_millis = 1
-
-        self._heartbeat_millis = heartbeat_millis
 
         runtime_options = temporalio.bridge.runtime.RuntimeOptions(
             telemetry=telemetry._to_bridge_config(),

temporalio/worker/_replayer.py

@@ -273,8 +273,12 @@ def on_eviction_hook(
                         ),
                     ),
                     nonsticky_to_sticky_poll_ratio=1,
+                    no_remote_activities=True,
                     task_types=temporalio.bridge.worker.WorkerTaskTypes(
-                        True, False, False, False
+                        enable_workflows=True,
+                        enable_local_activities=False,
+                        enable_remote_activities=False,
+                        enable_nexus=False,
                     ),
                     sticky_queue_schedule_to_start_timeout_millis=1000,
                     max_heartbeat_throttle_interval_millis=1000,

temporalio/worker/_worker.py

@@ -583,7 +583,8 @@ def check_activity(activity):
                 or not config["activities"],
                 task_types=temporalio.bridge.worker.WorkerTaskTypes(
                     enable_workflows=self._workflow_worker is not None,
-                    enable_local_activities=self._activity_worker is not None,
+                    enable_local_activities=self._activity_worker is not None
+                    and self._workflow_worker is not None,
                     enable_remote_activities=self._activity_worker is not None
                     and not config["no_remote_activities"],
                     enable_nexus=self._nexus_worker is not None,
@@ -885,6 +886,7 @@ class WorkerConfig(TypedDict, total=False):
     nexus_task_executor: Optional[concurrent.futures.Executor]
     workflow_runner: WorkflowRunner
     unsandboxed_workflow_runner: WorkflowRunner
+    plugins: Sequence[Plugin]
     interceptors: Sequence[Interceptor]
     build_id: Optional[str]
     identity: Optional[str]
@@ -915,7 +917,6 @@ class WorkerConfig(TypedDict, total=False):
     workflow_task_poller_behavior: PollerBehavior
     activity_task_poller_behavior: PollerBehavior
     nexus_task_poller_behavior: PollerBehavior
-    plugins: Sequence[Plugin]
 
 
 def _warn_if_activity_executor_max_workers_is_inconsistent(

tests/contrib/openai_agents/test_openai.py

@@ -2615,8 +2615,7 @@ async def test_split_workers(client: Client):
 
     # Workflow worker
     async with new_worker(
-        workflow_client,
-        HelloWorldAgent,
+        workflow_client, HelloWorldAgent, no_remote_activities=True
     ) as worker:
         activity_plugin = openai_agents.OpenAIAgentsPlugin(
             model_params=ModelActivityParameters(

tests/worker/test_workflow.py

@@ -5210,7 +5210,58 @@ async def run(self) -> None:
             await asyncio.sleep(0.1)
 
 
-async def test_workflow_replace_worker_client(client: Client):
+async def test_workflow_replace_worker_client(client: Client, env: WorkflowEnvironment):
+    if env.supports_time_skipping:
+        pytest.skip("Only testing against two real servers")
+    # We are going to start a second ephemeral server and then replace the
+    # client. So we will start a no-cache ticking workflow with the current
+    # client and confirm it has accomplished at least one task. Then we will
+    # start another on the other client, and confirm it gets started too. Then
+    # we will terminate both. We have to use a ticking workflow with only one
+    # poller to force a quick re-poll to recognize our client change quickly (as
+    # opposed to just waiting the minute for poll timeout).
+    async with await WorkflowEnvironment.start_local(
+        runtime=client.service_client.config.runtime,
+        dev_server_download_version=DEV_SERVER_DOWNLOAD_VERSION,
+    ) as other_env:
+        # Start both workflows on different servers
+        task_queue = f"tq-{uuid.uuid4()}"
+        handle1 = await client.start_workflow(
+            TickingWorkflow.run, id=f"wf-{uuid.uuid4()}", task_queue=task_queue
+        )
+        handle2 = await other_env.client.start_workflow(
+            TickingWorkflow.run, id=f"wf-{uuid.uuid4()}", task_queue=task_queue
+        )
+
+        async def any_task_completed(handle: WorkflowHandle) -> bool:
+            async for e in handle.fetch_history_events():
+                if e.HasField("workflow_task_completed_event_attributes"):
+                    return True
+            return False
+
+        # Now start the worker on the first env
+        async with Worker(
+            client,
+            task_queue=task_queue,
+            workflows=[TickingWorkflow],
+            max_cached_workflows=0,
+            max_concurrent_workflow_task_polls=1,
+        ) as worker:
+            # Confirm the first ticking workflow has completed a task but not
+            # the second
+            await assert_eq_eventually(True, lambda: any_task_completed(handle1))
+            assert not await any_task_completed(handle2)
+            # Now replace the client, which should be used fairly quickly
+            # because we should have timer-done poll completions every 100ms
+            worker.client = other_env.client
+            # Now confirm the other workflow has started
+            await assert_eq_eventually(True, lambda: any_task_completed(handle2))
+            # Terminate both
+            await handle1.terminate()
+            await handle2.terminate()
+
+
+async def test_workflow_replace_worker_client_diff_runtimes_fail(client: Client):
     other_runtime = Runtime(telemetry=TelemetryConfig())
     other_client = await Client.connect(
         client.service_client.config.target_host,