New test passing plugin names to core

Author: yuandrew

temporalio/bridge/sdk-core

@@ -124,10 +124,9 @@ pub fn init_runtime(options: RuntimeOptions) -> PyResult<RuntimeRef> {
 
     // Build metric config, but actual metrics instance is late-bound after
     // CoreRuntime is created since it needs Tokio runtime
-    let mut metrics_conf = metrics;
-    if let Some(metrics_conf_ref) = metrics_conf.as_ref() {
-        telemetry_build.attach_service_name(metrics_conf_ref.attach_service_name);
-        if let Some(prefix) = &metrics_conf_ref.metric_prefix {
+    if let Some(metrics_conf) = metrics.as_ref() {
+        telemetry_build.attach_service_name(metrics_conf.attach_service_name);
+        if let Some(prefix) = &metrics_conf.metric_prefix {
             telemetry_build.metric_prefix(prefix.to_string());
         }
     }
@@ -156,7 +155,7 @@ pub fn init_runtime(options: RuntimeOptions) -> PyResult<RuntimeRef> {
     // We late-bind the metrics after core runtime is created since it needs
     // the Tokio handle
     let mut metrics_call_buffer: Option<Arc<MetricsCallBuffer<BufferedMetricRef>>> = None;
-    if let Some(metrics_conf) = metrics_conf.take() {
+    if let Some(metrics_conf) = metrics {
         let _guard = core.tokio_handle().enter();
         // If they want buffered, cannot have Prom/OTel and we make buffered
         if metrics_conf.buffered_with_size > 0 {

temporalio/bridge/src/runtime.rs

@@ -647,11 +647,12 @@ impl WorkerRef {
         Ok(())
     }
 
-    fn replace_client(&self, client: &client::ClientRef) {
+    fn replace_client(&self, client: &client::ClientRef) -> PyResult<()> {
         self.worker
             .as_ref()
             .expect("missing worker")
-            .replace_client(client.retry_client.clone().into_inner());
+            .replace_client(client.retry_client.clone().into_inner())
+            .map_err(|err| PyValueError::new_err(format!("Failed replacing client: {err}")))
     }
 
     fn initiate_shutdown(&self) -> PyResult<()> {

temporalio/bridge/src/worker.rs

@@ -395,8 +395,6 @@ def _init_from_config(self, client: temporalio.client.Client, config: WorkerConf
         """
         self._config = config
 
-        config.setdefault("skip_client_worker_set_check", False)
-
         if not (
             config["activities"]
             or config["nexus_service_handlers"]

temporalio/worker/_worker.py

@@ -9,6 +9,7 @@
 import temporalio.client
 import temporalio.converter
 import temporalio.worker
+import temporalio.bridge.temporal_sdk_bridge
 from temporalio import workflow
 from temporalio.client import Client, ClientConfig, OutboundInterceptor, WorkflowHistory
 from temporalio.contrib.pydantic import pydantic_data_converter
@@ -173,6 +174,35 @@ async def test_worker_plugin_basic_config(client: Client) -> None:
     ]
 
 
+async def test_worker_plugin_names_forwarded_to_core(
+    client: Client, monkeypatch: pytest.MonkeyPatch
+) -> None:
+    captured_plugins: list[str] = []
+
+    original_new_worker = temporalio.bridge.temporal_sdk_bridge.new_worker
+
+    def new_worker_wrapper(runtime_ref, client_ref, config):
+        nonlocal captured_plugins
+        captured_plugins = list(config.plugins)
+        return original_new_worker(runtime_ref, client_ref, config)
+
+    monkeypatch.setattr(
+        temporalio.bridge.temporal_sdk_bridge,
+        "new_worker",
+        new_worker_wrapper,
+    )
+
+    plugin1 = SimplePlugin("test-worker-plugin1")
+    plugin2 = SimplePlugin("test-worker-plugin2")
+    worker = Worker(
+        client,
+        task_queue="queue",
+        activities=[never_run_activity],
+        plugins=[plugin1, plugin2],
+    )
+    assert captured_plugins == [plugin1.name(), plugin2.name()]
+
+
 async def test_worker_duplicated_plugin(client: Client) -> None:
     new_config = client.config()
     new_config["plugins"] = [MyCombinedPlugin()]

tests/test_plugins.py

@@ -1375,7 +1375,6 @@ async def _execute_workflow_with_activity(
     worker_config["activities"] = [fn] + additional_activities
     worker_config["shared_state_manager"] = _default_shared_state_manager
     worker_config["skip_client_worker_set_check"] = True
-    print("worker_config[skip_client_worker_set_check] = True\n")
     if not worker_config.get("max_concurrent_activities"):
         worker_config["max_concurrent_activities"] = default_max_concurrent_activities
     async with Worker(**worker_config):