rename

Author: brianstrauch

temporalio/contrib/openai_agents/README.md

@@ -593,7 +593,7 @@ External consumers (UIs, tracing pipelines, etc.) observe events as
 they arrive by hosting a [`WorkflowStream`](../workflow_streams/README.md)
 in the workflow and subscribing with `WorkflowStreamClient`. The
 streaming activity publishes each event to the topic configured on
-`ModelActivityParameters.streaming_event_topic`. The topic is required
+`ModelActivityParameters.streaming_topic`. The topic is required
 when using `Runner.run_streamed`; calling it without a configured topic
 raises before any activity is scheduled.
 
@@ -621,7 +621,7 @@ class MyAgent:
 
 To publish raw model events to external subscribers, host a
 `WorkflowStream` in the workflow and configure
-`OpenAIAgentsPlugin(model_params=ModelActivityParameters(streaming_event_topic="events"))`. See [`temporalio.contrib.workflow_streams`](../workflow_streams/README.md) for the
+`OpenAIAgentsPlugin(model_params=ModelActivityParameters(streaming_topic="events"))`. See [`temporalio.contrib.workflow_streams`](../workflow_streams/README.md) for the
 publisher and subscriber API.
 
 `RunResultStreaming.stream_events()` yields the agents-SDK

temporalio/contrib/openai_agents/_invoke_model_activity.py

@@ -195,8 +195,8 @@ class StreamingActivityModelInput(ActivityModelInput, total=False):
     Adds the streaming-only fields on top of :class:`ActivityModelInput`.
     """
 
-    streaming_event_topic: Required[str]
-    streaming_event_batch_interval: timedelta
+    streaming_topic: Required[str]
+    streaming_batch_interval: timedelta
 
 
 async def _empty_on_invoke_tool(_ctx: RunContextWrapper[Any], _input: str) -> str:
@@ -354,7 +354,7 @@ async def invoke_model_activity_streaming(
         terminal ``ResponseCompletedEvent``.
 
         Each event is also published to the workflow's stream on
-        ``streaming_event_topic`` so external consumers (UIs, tracing,
+        ``streaming_topic`` so external consumers (UIs, tracing,
         etc.) can observe events as they arrive.
 
         Heartbeats run on a background task via ``_auto_heartbeater`` so
@@ -364,9 +364,9 @@ async def invoke_model_activity_streaming(
         model = self._model_provider.get_model(input.get("model_name"))
         tools, handoffs = _build_tools_and_handoffs(input)
 
-        topic = input["streaming_event_topic"]
+        topic = input["streaming_topic"]
         batch_interval = input.get(
-            "streaming_event_batch_interval", timedelta(milliseconds=100)
+            "streaming_batch_interval", timedelta(milliseconds=100)
         )
         events: list[TResponseStreamEvent] = []
 

temporalio/contrib/openai_agents/_model_parameters.py

@@ -69,7 +69,7 @@ class ModelActivityParameters:
     use_local_activity: bool = False
     """Whether to use a local activity. If changed during a workflow execution, that would break determinism."""
 
-    streaming_event_topic: str | None = None
+    streaming_topic: str | None = None
     """Stream topic to publish raw model stream events to when the workflow
     calls ``Runner.run_streamed``. Required for ``Runner.run_streamed``;
     if left as ``None``, ``run_streamed`` raises before scheduling any
@@ -84,7 +84,7 @@ class ModelActivityParameters:
         Streaming support is experimental and may change in future
         versions."""
 
-    streaming_event_batch_interval: timedelta = timedelta(milliseconds=100)
+    streaming_batch_interval: timedelta = timedelta(milliseconds=100)
     """Interval between automatic flushes for the stream publisher used
     by the streaming activity.
 

temporalio/contrib/openai_agents/_openai_runner.py

@@ -268,10 +268,10 @@ def run_streamed(
         # captured into ``RunResultStreaming._stored_exception`` and may
         # be silently dropped if the queue completion sentinel is read
         # before the run_loop_task is observed as done.
-        if self.model_params.streaming_event_topic is None:
+        if self.model_params.streaming_topic is None:
             raise AgentsWorkflowError(
                 "Runner.run_streamed requires "
-                "ModelActivityParameters.streaming_event_topic to be set."
+                "ModelActivityParameters.streaming_topic to be set."
             )
         if self.model_params.use_local_activity:
             raise AgentsWorkflowError(

temporalio/contrib/openai_agents/_temporal_model_stub.py

@@ -245,11 +245,11 @@ async def stream_response(
                 "workflow stream signal channel)."
             )
 
-        topic = self.model_params.streaming_event_topic
+        topic = self.model_params.streaming_topic
         if topic is None:
             raise ValueError(
                 "Runner.run_streamed requires "
-                "ModelActivityParameters.streaming_event_topic to be set."
+                "ModelActivityParameters.streaming_topic to be set."
             )
 
         base_input, summary = self._build_activity_input(
@@ -266,10 +266,8 @@ async def stream_response(
         )
         streaming_input: StreamingActivityModelInput = {
             **base_input,
-            "streaming_event_topic": topic,
-            "streaming_event_batch_interval": (
-                self.model_params.streaming_event_batch_interval
-            ),
+            "streaming_topic": topic,
+            "streaming_batch_interval": (self.model_params.streaming_batch_interval),
         }
 
         events = await workflow.execute_activity_method(

tests/contrib/openai_agents/test_openai_streaming.py

@@ -205,7 +205,7 @@ def get_workflow_event_types(self) -> list[str]:
 @workflow.defn
 class StreamingRequiresTopicWorkflow:
     """Workflow that opts into ``Runner.run_streamed`` while the model
-    plugin was configured without a ``streaming_event_topic``.
+    plugin was configured without a ``streaming_topic``.
 
     The stub raises before scheduling the streaming activity; this
     propagates out of ``Runner.run_streamed`` and fails the workflow.
@@ -232,7 +232,7 @@ async def test_streaming_publishes_raw_events(client: Client):
         model=StreamingTestModel(),
         model_params=ModelActivityParameters(
             start_to_close_timeout=timedelta(seconds=30),
-            streaming_event_topic="events",
+            streaming_topic="events",
         ),
     ) as env:
         client = env.applied_on_client(client)
@@ -300,7 +300,7 @@ async def test_streaming_requires_topic(client: Client):
         model=StreamingTestModel(),
         model_params=ModelActivityParameters(
             start_to_close_timeout=timedelta(seconds=30),
-            streaming_event_topic=None,
+            streaming_topic=None,
         ),
     ) as env:
         client = env.applied_on_client(client)
@@ -316,7 +316,7 @@ async def test_streaming_requires_topic(client: Client):
                     execution_timeout=timedelta(seconds=30),
                 )
 
-    assert "streaming_event_topic" in str(exc_info.value.cause)
+    assert "streaming_topic" in str(exc_info.value.cause)
 
 
 @pytest.mark.asyncio
@@ -328,7 +328,7 @@ async def test_streaming_rejects_local_activity(client: Client):
         model=StreamingTestModel(),
         model_params=ModelActivityParameters(
             start_to_close_timeout=timedelta(seconds=30),
-            streaming_event_topic="events",
+            streaming_topic="events",
             use_local_activity=True,
         ),
     ) as env: