Plumb worker task types to lang

Author: yuandrew

temporalio/bridge/src/worker.rs

@@ -54,7 +54,7 @@ pub struct WorkerConfig {
     workflow_task_poller_behavior: PollerBehavior,
     nonsticky_to_sticky_poll_ratio: f32,
     activity_task_poller_behavior: PollerBehavior,
-    no_remote_activities: bool,
+    task_types: WorkerTaskTypes,
     sticky_queue_schedule_to_start_timeout_millis: u64,
     max_heartbeat_throttle_interval_millis: u64,
     default_heartbeat_throttle_interval_millis: u64,
@@ -175,6 +175,23 @@ pub struct ResourceBasedSlotSupplier {
     tuner_config: ResourceBasedTunerConfig,
 }
 
+#[derive(FromPyObject)]
+pub struct WorkerTaskTypes {
+     enable_workflows: bool,
+     enable_activities: bool,
+     enable_nexus: bool,
+}
+
+impl From<&WorkerTaskTypes> for temporalio_common::worker::WorkerTaskTypes {
+    fn from(t: &WorkerTaskTypes) -> Self {
+        Self {
+            enable_workflows: t.enable_workflows,
+            enable_activities: t.enable_activities,
+            enable_nexus: t.enable_nexus,
+        }
+    }
+}
+
 #[pyclass]
 pub struct SlotReserveCtx {
     #[pyo3(get)]
@@ -692,7 +709,7 @@ fn convert_worker_config(
         .tuner(Arc::new(converted_tuner))
         .nonsticky_to_sticky_poll_ratio(conf.nonsticky_to_sticky_poll_ratio)
         .activity_task_poller_behavior(conf.activity_task_poller_behavior)
-        .no_remote_activities(conf.no_remote_activities)
+        .task_types(&conf.task_types)
         .sticky_queue_schedule_to_start_timeout(Duration::from_millis(
             conf.sticky_queue_schedule_to_start_timeout_millis,
         ))

temporalio/bridge/worker.py

@@ -54,7 +54,7 @@ class WorkerConfig:
     workflow_task_poller_behavior: PollerBehavior
     nonsticky_to_sticky_poll_ratio: float
     activity_task_poller_behavior: PollerBehavior
-    no_remote_activities: bool
+    task_types: WorkerTaskTypes
     sticky_queue_schedule_to_start_timeout_millis: int
     max_heartbeat_throttle_interval_millis: int
     default_heartbeat_throttle_interval_millis: int
@@ -170,6 +170,15 @@ class TunerHolder:
     nexus_slot_supplier: SlotSupplier
 
 
+@dataclass
+class WorkerTaskTypes:
+    """Python representation of the Rust struct for worker task types"""
+
+    enable_workflows: bool
+    enable_activities: bool
+    enable_nexus: bool
+
+
 class Worker:
     """SDK Core worker."""
 

temporalio/runtime.py

@@ -140,7 +140,15 @@ def formatted(self) -> str:
         """Return a formatted form of this filter."""
         # We intentionally aren't using __str__ or __format__ so they can keep
         # their original dataclass impls
-        return f"{self.other_level},temporalio_sdk_core={self.core_level},temporalio_client={self.core_level},temporalio_sdk={self.core_level}"
+        targets = [
+            "temporal_sdk_core",
+            "temporal_sdk_bridge",
+            "temporal_client",
+            "temporalio_sdk",
+        ]
+        parts = [self.other_level]
+        parts.extend(f"{target}={self.core_level}" for target in targets)
+        return ",".join(parts)
 
 
 @dataclass(frozen=True)

temporalio/worker/_replayer.py

@@ -9,6 +9,7 @@
 from dataclasses import dataclass
 from typing import AsyncIterator, Dict, Mapping, Optional, Sequence, Type
 
+from temporalio.bridge.worker import WorkerTaskTypes
 from typing_extensions import TypedDict
 
 import temporalio.api.history.v1
@@ -273,7 +274,7 @@ def on_eviction_hook(
                         ),
                     ),
                     nonsticky_to_sticky_poll_ratio=1,
-                    no_remote_activities=True,
+                    task_types=WorkerTaskTypes(True, False, False),
                     sticky_queue_schedule_to_start_timeout_millis=1000,
                     max_heartbeat_throttle_interval_millis=1000,
                     default_heartbeat_throttle_interval_millis=1000,

temporalio/worker/_worker.py

@@ -22,6 +22,7 @@
     cast,
 )
 
+from temporalio.bridge.worker import WorkerTaskTypes
 from typing_extensions import TypeAlias, TypedDict
 
 import temporalio.bridge.worker
@@ -124,7 +125,6 @@ def __init__(
         max_concurrent_workflow_task_polls: Optional[int] = None,
         nonsticky_to_sticky_poll_ratio: float = 0.2,
         max_concurrent_activity_task_polls: Optional[int] = None,
-        no_remote_activities: bool = False,
         sticky_queue_schedule_to_start_timeout: timedelta = timedelta(seconds=10),
         max_heartbeat_throttle_interval: timedelta = timedelta(seconds=60),
         default_heartbeat_throttle_interval: timedelta = timedelta(seconds=30),
@@ -251,8 +251,6 @@ def __init__(
                 If set, will override any value passed to ``activity_task_poller_behavior``.
 
                 WARNING: Deprecated, use ``activity_task_poller_behavior`` instead
-            no_remote_activities: If true, this worker will only handle workflow
-                tasks and local activities, it will not poll for activity tasks.
             sticky_queue_schedule_to_start_timeout: How long a workflow task is
                 allowed to sit on the sticky queue before it is timed out and
                 moved to the non-sticky queue where it may be picked up by any
@@ -346,7 +344,6 @@ def __init__(
             max_concurrent_workflow_task_polls=max_concurrent_workflow_task_polls,
             nonsticky_to_sticky_poll_ratio=nonsticky_to_sticky_poll_ratio,
             max_concurrent_activity_task_polls=max_concurrent_activity_task_polls,
-            no_remote_activities=no_remote_activities,
             sticky_queue_schedule_to_start_timeout=sticky_queue_schedule_to_start_timeout,
             max_heartbeat_throttle_interval=max_heartbeat_throttle_interval,
             default_heartbeat_throttle_interval=default_heartbeat_throttle_interval,
@@ -576,11 +573,11 @@ def check_activity(activity):
                 max_cached_workflows=config["max_cached_workflows"],
                 tuner=bridge_tuner,
                 nonsticky_to_sticky_poll_ratio=config["nonsticky_to_sticky_poll_ratio"],
-                # We have to disable remote activities if a user asks _or_ if we
-                # are not running an activity worker at all. Otherwise shutdown
-                # will not proceed properly.
-                no_remote_activities=config["no_remote_activities"]
-                or not config["activities"],
+                task_types=WorkerTaskTypes(
+                    enable_workflows=self._workflow_worker is not None,
+                    enable_activities=self._activity_worker is not None,
+                    enable_nexus=self._nexus_worker is not None,
+                ),
                 sticky_queue_schedule_to_start_timeout_millis=int(
                     1000
                     * config["sticky_queue_schedule_to_start_timeout"].total_seconds()
@@ -890,7 +887,6 @@ class WorkerConfig(TypedDict, total=False):
     max_concurrent_workflow_task_polls: Optional[int]
     nonsticky_to_sticky_poll_ratio: float
     max_concurrent_activity_task_polls: Optional[int]
-    no_remote_activities: bool
     sticky_queue_schedule_to_start_timeout: timedelta
     max_heartbeat_throttle_interval: timedelta
     default_heartbeat_throttle_interval: timedelta

tests/contrib/openai_agents/test_openai.py

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