Readme fixes, move plugin to its own file

Author: tconley1428

README.md

@@ -1578,19 +1578,20 @@ class MonitoringPlugin(Plugin):
             self.logger.info("Worker execution completed")
 
   def configure_replayer(self, config: ReplayerConfig) -> ReplayerConfig:
-      config["data_converter"] = pydantic_data_converter
-      config["workflows"] = list(config.get("workflows") or []) + [HelloWorkflow]
-      return config
+      return self.next_worker_plugin.configure_replayer(config)
 
   @asynccontextmanager
   async def workflow_replay(
       self,
       replayer: Replayer,
       histories: AsyncIterator[temporalio.client.WorkflowHistory],
   ) -> AsyncIterator[AsyncIterator[WorkflowReplayResult]]:
-      with set_some_context():
-          async with super().workflow_replay(replayer, histories) as results:
+        self.logger.info("Starting replay execution")
+        try:
+          async with self.next_worker_plugin.workflow_replay(replayer, histories) as results:
               yield results
+        finally:
+            self.logger.info("Replay execution completed")
 
 # Use the plugin when creating a worker
 worker = Worker(
@@ -1606,69 +1607,64 @@ For plugins that need to work with both clients and workers, you can implement b
 
 ```python
 import temporalio
-from contextlib import asynccontextmanager
+from contextlib import AbstractAsyncContextManager
 from typing import AsyncIterator
 from temporalio.client import Plugin as ClientPlugin, ClientConfig
 from temporalio.worker import Plugin as WorkerPlugin, WorkerConfig, ReplayerConfig, Worker, Replayer, WorkflowReplayResult
 
 
 class UnifiedPlugin(ClientPlugin, WorkerPlugin):
-  def init_client_plugin(self, next: ClientPlugin) -> ClientPlugin:
-    self.next_client_plugin = next
-    return self
+    def init_client_plugin(self, next: ClientPlugin) -> ClientPlugin:
+        self.next_client_plugin = next
+        return self
 
-  def init_worker_plugin(self, next: WorkerPlugin) -> WorkerPlugin:
-    self.next_worker_plugin = next
-    return self
+    def init_worker_plugin(self, next: WorkerPlugin) -> WorkerPlugin:
+        self.next_worker_plugin = next
+        return self
 
-  def configure_client(self, config: ClientConfig) -> ClientConfig:
-    # Client-side customization
-    config["namespace"] = "unified-namespace"
-    return super().configure_client(config)
-
-  async def connect_service_client(
-      self, config: temporalio.service.ConnectConfig
-  ) -> temporalio.service.ServiceClient:
-      # Add authentication to the connection
-      config.api_key = self.api_key
-      return await self.next_client_plugin.connect_service_client(config)
-      
-  def configure_worker(self, config: WorkerConfig) -> WorkerConfig:
-    # Worker-side customization
-    config["max_cached_workflows"] = 500
-    return super().configure_worker(config)
-
-  async def run_worker(self, worker: Worker) -> None:
-    print("Starting unified worker")
-    await super().run_worker(worker)
-      
-  def configure_replayer(self, config: ReplayerConfig) -> ReplayerConfig:
-      config["data_converter"] = pydantic_data_converter
-      config["workflows"] = list(config.get("workflows") or []) + [HelloWorkflow]
-      return config
+    def configure_client(self, config: ClientConfig) -> ClientConfig:
+        # Client-side customization
+        config["data_converter"] = pydantic_data_converter
+        return super().configure_client(config)
 
-  @asynccontextmanager
-  async def workflow_replay(
-      self,
-      replayer: Replayer,
-      histories: AsyncIterator[temporalio.client.WorkflowHistory],
-  ) -> AsyncIterator[AsyncIterator[WorkflowReplayResult]]:
-      with set_some_context():
-          async with super().workflow_replay(replayer, histories) as results:
-              yield results
+    async def connect_service_client(
+        self, config: temporalio.service.ConnectConfig
+    ) -> temporalio.service.ServiceClient:
+        # Add authentication to the connection
+        config.api_key = self.api_key
+        return await self.next_client_plugin.connect_service_client(config)
+        
+    def configure_worker(self, config: WorkerConfig) -> WorkerConfig:
+        # Worker-side customization
+        return super().configure_worker(config)
+  
+    async def run_worker(self, worker: Worker) -> None:
+        print("Starting unified worker")
+        await super().run_worker(worker)
+        
+    def configure_replayer(self, config: ReplayerConfig) -> ReplayerConfig:
+        config["data_converter"] = pydantic_data_converter
+        return config
+    
+    async def workflow_replay(
+        self,
+        replayer: Replayer,
+        histories: AsyncIterator[temporalio.client.WorkflowHistory],
+    ) -> AbstractAsyncContextManager[AsyncIterator[WorkflowReplayResult]]:
+        return self.next_worker_plugin.workflow_replay(replayer, histories)
 
 # Create client with the unified plugin
 client = await Client.connect(
-  "localhost:7233",
-  plugins=[UnifiedPlugin()]
+    "localhost:7233",
+    plugins=[UnifiedPlugin()]
 )
 
 # Worker will automatically inherit the plugin from the client
 worker = Worker(
-  client,
-  task_queue="my-task-queue",
-  workflows=[MyWorkflow],
-  activities=[my_activity]
+    client,
+    task_queue="my-task-queue",
+    workflows=[MyWorkflow],
+    activities=[my_activity]
 )
 ```
 

temporalio/worker/__init__.py

@@ -21,6 +21,7 @@
     WorkflowInterceptorClassInput,
     WorkflowOutboundInterceptor,
 )
+from ._plugin import Plugin
 from ._replayer import (
     Replayer,
     ReplayerConfig,
@@ -44,7 +45,6 @@
     WorkflowSlotInfo,
 )
 from ._worker import (
-    Plugin,
     PollerBehavior,
     PollerBehaviorAutoscaling,
     PollerBehaviorSimpleMaximum,

temporalio/worker/_plugin.py

@@ -0,0 +1,121 @@
+from __future__ import annotations
+
+import abc
+from contextlib import AbstractAsyncContextManager
+from typing import TYPE_CHECKING, AsyncIterator
+
+from temporalio.client import WorkflowHistory
+
+if TYPE_CHECKING:
+    from temporalio.worker import (
+        Replayer,
+        ReplayerConfig,
+        Worker,
+        WorkerConfig,
+        WorkflowReplayResult,
+    )
+
+
+class Plugin(abc.ABC):
+    """Base class for worker plugins that can intercept and modify worker behavior.
+
+    Plugins allow customization of worker creation and execution processes
+    through a chain of responsibility pattern. Each plugin can modify the worker
+    configuration or intercept worker execution.
+
+    WARNING: This is an experimental feature and may change in the future.
+    """
+
+    def name(self) -> str:
+        """Get the qualified name of this plugin. Can be overridden if desired to provide a more appropriate name.
+
+        Returns:
+            The fully qualified name of the plugin class (module.classname).
+        """
+        return type(self).__module__ + "." + type(self).__qualname__
+
+    @abc.abstractmethod
+    def init_worker_plugin(self, next: Plugin) -> Plugin:
+        """Initialize this plugin in the plugin chain.
+
+        This method sets up the chain of responsibility pattern by storing a reference
+        to the next plugin in the chain. It is called during worker creation to build
+        the plugin chain.
+
+        Args:
+            next: The next plugin in the chain to delegate to.
+
+        Returns:
+            This plugin instance for method chaining.
+        """
+
+    @abc.abstractmethod
+    def configure_worker(self, config: WorkerConfig) -> WorkerConfig:
+        """Hook called when creating a worker to allow modification of configuration.
+
+        This method is called during worker creation and allows plugins to modify
+        the worker configuration before the worker is fully initialized. Plugins
+        can modify task queue names, adjust concurrency settings, add interceptors,
+        or change other worker settings.
+
+        Args:
+            config: The worker configuration dictionary to potentially modify.
+
+        Returns:
+            The modified worker configuration.
+        """
+
+    @abc.abstractmethod
+    async def run_worker(self, worker: Worker) -> None:
+        """Hook called when running a worker to allow interception of execution.
+
+        This method is called when the worker is started and allows plugins to
+        intercept or wrap the worker execution. Plugins can add monitoring,
+        custom lifecycle management, or other execution-time behavior.
+
+        Args:
+            worker: The worker instance to run.
+        """
+
+    @abc.abstractmethod
+    def configure_replayer(self, config: ReplayerConfig) -> ReplayerConfig:
+        """Hook called when creating a replayer to allow modification of configuration.
+
+        This should be used to configure anything in ReplayerConfig needed to make execution match
+        the original. This could include interceptors, DataConverter, workflows, and more.
+
+        Args:
+            config: The replayer configuration dictionary to potentially modify.
+
+        Returns:
+            The modified replayer configuration.
+        """
+
+    @abc.abstractmethod
+    def workflow_replay(
+        self,
+        replayer: Replayer,
+        histories: AsyncIterator[WorkflowHistory],
+    ) -> AbstractAsyncContextManager[AsyncIterator[WorkflowReplayResult]]:
+        """Hook called when running a replayer to allow interception of execution."""
+
+
+class _RootPlugin(Plugin):
+    def init_worker_plugin(self, next: Plugin) -> Plugin:
+        raise NotImplementedError()
+
+    def configure_worker(self, config: WorkerConfig) -> WorkerConfig:
+        return config
+
+    def configure_replayer(self, config: ReplayerConfig) -> ReplayerConfig:
+        return config
+
+    async def run_worker(self, worker: Worker) -> None:
+        await worker._run()
+
+    def workflow_replay(
+        self,
+        replayer: Replayer,
+        histories: AsyncIterator[WorkflowHistory],
+    ) -> AbstractAsyncContextManager[AsyncIterator[WorkflowReplayResult]]:
+        return replayer._workflow_replay_iterator(histories)

temporalio/worker/_replayer.py

@@ -21,6 +21,8 @@
 
 from ..common import HeaderCodecBehavior
 from ._interceptor import Interceptor
+from ._plugin import _RootPlugin
+from ._worker import load_default_build_id
 from ._workflow import _WorkflowWorker
 from ._workflow_instance import UnsandboxedWorkflowRunner, WorkflowRunner
 from .workflow_sandbox import SandboxedWorkflowRunner
@@ -81,8 +83,6 @@ def __init__(
             header_codec_behavior=header_codec_behavior,
         )
 
-        from ._worker import _RootPlugin
-
         # Apply plugin configuration
         root_plugin: temporalio.worker.Plugin = _RootPlugin()
         for plugin in reversed(plugins):
@@ -240,8 +240,6 @@ def on_eviction_hook(
                 != HeaderCodecBehavior.NO_CODEC,
             )
             # Create bridge worker
-            from ._worker import load_default_build_id
-
             bridge_worker, pusher = temporalio.bridge.worker.Worker.for_replay(
                 runtime._core_runtime,
                 temporalio.bridge.worker.WorkerConfig(