workflow_streams: remove WorkflowStream(Client).publish in favor of topic handles

Drops the un-typed publish(topic, value) entry points on
WorkflowStream and WorkflowStreamClient. Publishers now go through
WorkflowStream.topic(name, type=T) and
WorkflowStreamClient.topic(name, type=T) which return typed
TopicHandle / WorkflowTopicHandle objects.

Topic handles are the only supported publish API. Per Codex
review on PR #1423: per-instance type uniqueness is a
factory-level invariant — handle-based publishing on a single
publisher cannot mix Ts on a topic, while still allowing escape
hatches (type=typing.Any for heterogeneous topics, pre-built
Payload values via the zero-copy fast path on any-typed handle).

Migrates the internal users:
- temporalio/contrib/openai_agents/_invoke_model_activity.py uses
  type=Any (TResponseStreamEvent is an annotated union, not a class)
- temporalio/contrib/google_adk_agents/_model.py uses
  type=LlmResponse
- All workflow_streams tests migrated to the handle form,
  preserving existing topic names and types

Author: jssmith

temporalio/contrib/google_adk_agents/_model.py

@@ -79,13 +79,14 @@ async def invoke_model_streaming(
     stream = WorkflowStreamClient.from_within_activity(
         batch_interval=input.streaming_event_batch_interval,
     )
+    events = stream.topic(input.streaming_event_topic, type=LlmResponse)
     async with stream:
         async for response in llm.generate_content_async(
             llm_request=llm_request, stream=True
         ):
             activity.heartbeat()
             responses.append(response)
-            stream.publish(input.streaming_event_topic, response)
+            events.publish(response)
 
     return responses
 

temporalio/contrib/openai_agents/_invoke_model_activity.py

@@ -6,7 +6,7 @@
 import enum
 from dataclasses import dataclass
 from datetime import timedelta
-from typing import Any, NoReturn
+from typing import Any, NoReturn, cast
 
 from agents import (
     AgentOutputSchemaBase,
@@ -373,6 +373,11 @@ async def invoke_model_activity_streaming(
         stream = WorkflowStreamClient.from_within_activity(
             batch_interval=batch_interval
         )
+        # TResponseStreamEvent is a typing.Annotated[Union[...]] — a typing
+        # special form, not a class — so it cannot be passed as type[T].
+        # Use Any here; subscribers that want typed decode can pass
+        # result_type=TResponseStreamEvent on their own subscribe call.
+        events_topic = stream.topic(topic, type=cast("type[Any]", cast(object, Any)))
         async with stream:
             try:
                 async for event in model.stream_response(
@@ -388,7 +393,7 @@ async def invoke_model_activity_streaming(
                     prompt=input.get("prompt"),
                 ):
                     events.append(event)
-                    stream.publish(topic, event)
+                    events_topic.publish(event)
             except APIStatusError as e:
                 _raise_for_openai_status(e)
 

temporalio/contrib/workflow_streams/_client.py

@@ -62,13 +62,14 @@ class WorkflowStreamClient:
     :py:meth:`from_within_activity` (infer both from the current activity
     context), or by passing a handle directly to the constructor.
 
-    For publishing, use as an async context manager to get automatic
-    batching::
+    For publishing, bind a typed topic handle and use the client as
+    an async context manager to get automatic batching::
 
         client = WorkflowStreamClient.create(temporal_client, workflow_id)
+        events = client.topic("events", type=MyEvent)
         async with client:
-            client.publish("events", my_event)
-            client.publish("events", another_event, force_flush=True)
+            events.publish(my_event)
+            events.publish(another_event, force_flush=True)
             ...  # more publishing
         # Buffer is flushed automatically on context manager exit.
 
@@ -230,34 +231,14 @@ async def __aexit__(self, *_exc: object) -> None:
         while self._pending is not None or self._buffer:
             await self._flush()
 
-    def publish(self, topic: str, value: Any, force_flush: bool = False) -> None:
-        """Buffer a message for publishing.
-
-        .. deprecated::
-            Prefer :meth:`topic` and :meth:`TopicHandle.publish`. The
-            handle form carries the value type, which is needed for
-            cross-language SDK consistency.
-
-        ``value`` may be any Python value the client's payload
-        converter can handle, or a pre-built
-        :class:`temporalio.api.common.v1.Payload` for zero-copy. The
-        codec chain is not applied per item — it runs once on the
-        signal envelope that delivers the batch.
-
-        Args:
-            topic: Topic string.
-            value: Value to publish. Converted to a ``Payload`` via
-                the client's sync payload converter at flush time.
-                Pre-built ``Payload`` instances bypass conversion.
-            force_flush: If True, wake the flusher to send immediately
-                (fire-and-forget — does not block the caller).
-        """
-        self._publish_to_topic(topic, value, force_flush=force_flush)
-
     def _publish_to_topic(
         self, topic: str, value: Any, *, force_flush: bool = False
     ) -> None:
-        """Internal publish path shared by :meth:`publish` and topic handles."""
+        """Internal publish path used by :class:`TopicHandle`.
+
+        Not part of the public API — call
+        :meth:`TopicHandle.publish` instead.
+        """
         self._buffer.append((topic, value))
         if force_flush or (
             self._max_batch_size is not None

temporalio/contrib/workflow_streams/_stream.py

@@ -156,26 +156,12 @@ def __init__(self, prior_state: WorkflowStreamState | None = None) -> None:
         )
         workflow.set_query_handler(_OFFSET_QUERY, self._on_offset)
 
-    def publish(self, topic: str, value: Any) -> None:
-        """Publish an item from within workflow code.
-
-        .. deprecated::
-            Prefer :meth:`topic` and :meth:`WorkflowTopicHandle.publish`.
-            The handle form carries the value type, which is needed
-            for cross-language SDK consistency.
-
-        ``value`` may be any Python value the workflow's payload
-        converter can handle, or a pre-built
-        :class:`temporalio.api.common.v1.Payload` for zero-copy.
+    def _publish_to_topic(self, topic: str, value: Any) -> None:
+        """Internal publish path used by :class:`WorkflowTopicHandle`.
 
-        The codec chain is not applied here (it runs on the
-        ``__temporal_workflow_stream_poll`` update envelope that later
-        delivers the item to a subscriber).
+        Not part of the public API — call
+        :meth:`WorkflowTopicHandle.publish` instead.
         """
-        self._publish_to_topic(topic, value)
-
-    def _publish_to_topic(self, topic: str, value: Any) -> None:
-        """Internal publish path shared by :meth:`publish` and topic handles."""
         if isinstance(value, Payload):
             payload = value
         else:

tests/contrib/workflow_streams/test_workflow_streams.py

@@ -101,7 +101,7 @@ async def run(self, count: int) -> None:
             start_to_close_timeout=timedelta(seconds=30),
             heartbeat_timeout=timedelta(seconds=10),
         )
-        self.stream.publish("status", b"activity_done")
+        self.stream.topic("status", type=bytes).publish(b"activity_done")
         await workflow.wait_condition(lambda: self._closed)
 
 
@@ -125,7 +125,9 @@ def close(self) -> None:
     @workflow.run
     async def run(self, count: int) -> None:
         for i in range(count):
-            self.stream.publish("events", AgentEvent(kind="tick", payload={"i": i}))
+            self.stream.topic("events", type=AgentEvent).publish(
+                AgentEvent(kind="tick", payload={"i": i})
+            )
         await workflow.wait_condition(lambda: self._closed)
 
 
@@ -164,7 +166,7 @@ def close(self) -> None:
     @workflow.run
     async def run(self, count: int) -> None:
         for i in range(count):
-            self.stream.publish("events", f"item-{i}".encode())
+            self.stream.topic("events", type=bytes).publish(f"item-{i}".encode())
         await workflow.wait_condition(lambda: self._closed)
 
 
@@ -203,14 +205,14 @@ def close(self) -> None:
 
     @workflow.run
     async def run(self, count: int) -> None:
-        self.stream.publish("status", b"started")
+        self.stream.topic("status", type=bytes).publish(b"started")
         await workflow.execute_activity(
             "publish_items",
             count,
             start_to_close_timeout=timedelta(seconds=30),
             heartbeat_timeout=timedelta(seconds=10),
         )
-        self.stream.publish("status", b"done")
+        self.stream.topic("status", type=bytes).publish(b"done")
         await workflow.wait_condition(lambda: self._closed)
 
 
@@ -280,7 +282,7 @@ async def run(self, count: int) -> None:
             start_to_close_timeout=timedelta(seconds=30),
             heartbeat_timeout=timedelta(seconds=10),
         )
-        self.stream.publish("status", b"activity_done")
+        self.stream.topic("status", type=bytes).publish(b"activity_done")
         await workflow.wait_condition(lambda: self._closed)
 
 
@@ -349,7 +351,7 @@ async def publish_items(count: int) -> None:
     async with client:
         for i in range(count):
             activity.heartbeat()
-            client.publish("events", f"item-{i}".encode())
+            client.topic("events", type=bytes).publish(f"item-{i}".encode())
 
 
 @activity.defn(name="publish_multi_topic")
@@ -362,7 +364,7 @@ async def publish_multi_topic(count: int) -> None:
         for i in range(count):
             activity.heartbeat()
             topic = topics[i % len(topics)]
-            client.publish(topic, f"{topic}-{i}".encode())
+            client.topic(topic, type=bytes).publish(f"{topic}-{i}".encode())
 
 
 @activity.defn(name="publish_with_priority")
@@ -376,9 +378,9 @@ async def publish_with_priority() -> None:
         batch_interval=timedelta(seconds=60)
     )
     async with client:
-        client.publish("events", b"normal-0")
-        client.publish("events", b"normal-1")
-        client.publish("events", b"priority", force_flush=True)
+        client.topic("events", type=bytes).publish(b"normal-0")
+        client.topic("events", type=bytes).publish(b"normal-1")
+        client.topic("events", type=bytes).publish(b"priority", force_flush=True)
         for _ in range(100):
             activity.heartbeat()
             await asyncio.sleep(0.1)
@@ -392,7 +394,7 @@ async def publish_batch_test(count: int) -> None:
     async with client:
         for i in range(count):
             activity.heartbeat()
-            client.publish("events", f"item-{i}".encode())
+            client.topic("events", type=bytes).publish(f"item-{i}".encode())
 
 
 @activity.defn(name="publish_with_max_batch")
@@ -403,7 +405,7 @@ async def publish_with_max_batch(count: int) -> None:
     async with client:
         for i in range(count):
             activity.heartbeat()
-            client.publish("events", f"item-{i}".encode())
+            client.topic("events", type=bytes).publish(f"item-{i}".encode())
             # Yield so the flusher task can run when max_batch_size triggers
             # _flush_event. Real workloads (e.g. agents awaiting LLM streams)
             # yield constantly; a tight loop with no awaits would never let
@@ -1085,9 +1087,9 @@ async def test_explicit_flush_barrier(client: Client) -> None:
 
             # 2. Flush makes prior publishes visible without waiting on
             # the 60s batch timer.
-            stream.publish("events", b"a")
-            stream.publish("events", b"b")
-            stream.publish("events", b"c")
+            stream.topic("events", type=bytes).publish(b"a")
+            stream.topic("events", type=bytes).publish(b"b")
+            stream.topic("events", type=bytes).publish(b"c")
             await stream.flush()
             assert await stream.get_offset() == 3
 
@@ -1279,14 +1281,14 @@ async def maybe_failing_signal(*args: Any, **kwargs: Any) -> Any:
             return await real_signal(*args, **kwargs)
 
         with patch.object(handle, "signal", side_effect=maybe_failing_signal):
-            stream.publish("events", b"item-0")
-            stream.publish("events", b"item-1")
+            stream.topic("events", type=bytes).publish(b"item-0")
+            stream.topic("events", type=bytes).publish(b"item-1")
             with pytest.raises(RuntimeError):
                 await stream._flush()
 
             # Publish more during the failed state — must not overtake the
             # pending retry on eventual delivery.
-            stream.publish("events", b"item-2")
+            stream.topic("events", type=bytes).publish(b"item-2")
             with pytest.raises(RuntimeError):
                 await stream._flush()
 
@@ -1336,7 +1338,7 @@ async def maybe_failing_signal(*args: Any, **kwargs: Any) -> Any:
             ),
             patch.object(handle, "signal", side_effect=maybe_failing_signal),
         ):
-            stream.publish("events", b"lost")
+            stream.topic("events", type=bytes).publish(b"lost")
 
             # First flush fails and enters the pending-retry state.
             with pytest.raises(RuntimeError):
@@ -1351,7 +1353,7 @@ async def maybe_failing_signal(*args: Any, **kwargs: Any) -> Any:
 
             # Stop failing signals; subsequent publishes must succeed.
             fail_signals = False
-            stream.publish("events", b"kept")
+            stream.topic("events", type=bytes).publish(b"kept")
             await stream._flush()
 
         items = await collect_items(client, handle, None, 0, 1)
@@ -1586,7 +1588,7 @@ def __init__(self, prepub_count: int = 0) -> None:
         self.stream = WorkflowStream()
         self._closed = False
         for i in range(prepub_count):
-            self.stream.publish("events", f"item-{i}".encode())
+            self.stream.topic("events", type=bytes).publish(f"item-{i}".encode())
 
     @workflow.signal
     def close(self) -> None:
@@ -1890,7 +1892,7 @@ def close(self) -> None:
     @workflow.run
     async def run(self, count: int) -> None:
         for i in range(count):
-            self.stream.publish("events", f"broker-{i}".encode())
+            self.stream.topic("events", type=bytes).publish(f"broker-{i}".encode())
         await workflow.wait_condition(lambda: self._closed)
 
 
@@ -2001,7 +2003,7 @@ async def standalone_publish_to_broker(input: StandalonePublishInput) -> None:
     async with client:
         for i in range(input.count):
             activity.heartbeat()
-            client.publish("events", f"standalone-{i}".encode())
+            client.topic("events", type=bytes).publish(f"standalone-{i}".encode())
 
 
 @activity.defn(name="standalone_subscribe_to_broker")
@@ -2187,7 +2189,7 @@ def close(self) -> None:
     @workflow.run
     async def run(self, count: int) -> str:
         for i in range(count):
-            self.stream.publish("events", f"nexus-{i}".encode())
+            self.stream.topic("events", type=bytes).publish(f"nexus-{i}".encode())
         await workflow.wait_condition(lambda: self._closed)
         return "done"