PR Feedback

Author: tconley1428

temporalio/contrib/openai_agents/_invoke_model_activity.py

@@ -155,8 +155,9 @@ class ModelActivity:
 
     def __init__(self, model_provider: Optional[ModelProvider] = None):
         """Initialize the activity with a model provider."""
-        self._model_provider = model_provider
-        if model_provider is None and not workflow.in_workflow():
+        if model_provider:
+            self._model_provider = model_provider
+        else:
             self._model_provider = OpenAIProvider(
                 openai_client=AsyncOpenAI(max_retries=0)
             )
@@ -165,10 +166,6 @@ def __init__(self, model_provider: Optional[ModelProvider] = None):
     @_auto_heartbeater
     async def invoke_model_activity(self, input: ActivityModelInput) -> ModelResponse:
         """Activity that invokes a model with the given input."""
-        if not self._model_provider:
-            self._model_provider = OpenAIProvider(
-                openai_client=AsyncOpenAI(max_retries=0)
-            )
         model = self._model_provider.get_model(input.get("model_name"))
 
         async def empty_on_invoke_tool(ctx: RunContextWrapper[Any], input: str) -> str:

temporalio/plugin.py

@@ -86,8 +86,8 @@ def __init__(
                 Applied to the Worker and Replayer.
             workflow_failure_exception_types: Exception types for workflow failures to append,
                 or callable to customize existing ones. Applied to the Worker and Replayer.
-            run_context: Optional async context manager producer to wrap worker/replayer execution.
-                Applied to the Worker and Replayer.
+            run_context: A place to run custom code to wrap around the Worker (or Replayer) execution.
+                Specifically, it's an async context manager producer. Applied to the Worker and Replayer.
 
         Returns:
             A configured Plugin instance.

tests/contrib/openai_agents/test_openai_replay.py

@@ -1,6 +1,8 @@
 from pathlib import Path
 
 import pytest
+from agents import OpenAIProvider
+from openai import AsyncOpenAI
 
 from temporalio.client import WorkflowHistory
 from temporalio.contrib.openai_agents import ModelActivityParameters, OpenAIAgentsPlugin
@@ -42,5 +44,11 @@ async def test_replay(file_name: str) -> None:
                 InputGuardrailWorkflow,
                 OutputGuardrailWorkflow,
             ],
-            plugins=[OpenAIAgentsPlugin()],
+            plugins=[
+                OpenAIAgentsPlugin(
+                    model_provider=OpenAIProvider(
+                        openai_client=AsyncOpenAI(max_retries=0, api_key="PLACEHOLDER")
+                    )
+                )
+            ],
         ).replay_workflow(WorkflowHistory.from_json("fake", history_json))

tests/test_plugins.py

@@ -275,7 +275,7 @@ async def test_replay(client: Client) -> None:
     await replayer.replay_workflow(await handle.fetch_history())
 
 
-async def test_static_plugins(client: Client) -> None:
+async def test_simple_plugins(client: Client) -> None:
     plugin = SimplePlugin(
         "MyPlugin",
         data_converter=pydantic_data_converter,
@@ -311,7 +311,7 @@ async def test_static_plugins(client: Client) -> None:
     assert replayer.config().get("workflows") == [HelloWorkflow, HelloWorkflow2]
 
 
-async def test_static_plugins_callables(client: Client) -> None:
+async def test_simple_plugins_callables(client: Client) -> None:
     def converter(old: Optional[DataConverter]):
         if old != temporalio.converter.default():
             raise ValueError("Can't override non-default converter")
@@ -344,3 +344,21 @@ def converter(old: Optional[DataConverter]):
         plugins=[plugin],
     )
     assert worker.config().get("workflows") == []
+
+
+class MediumPlugin(SimplePlugin):
+    def __init__(self):
+        super().__init__("MediumPlugin", data_converter=pydantic_data_converter)
+
+    def configure_worker(self, config: WorkerConfig) -> WorkerConfig:
+        config = super().configure_worker(config)
+        config["task_queue"] = "override"
+        return config
+
+
+async def test_medium_plugin(client: Client) -> None:
+    plugin = MediumPlugin()
+    worker = Worker(
+        client, task_queue="queue", plugins=[plugin], workflows=[HelloWorkflow]
+    )
+    assert worker.config().get("task_queue") == "override"