samples: rename workflow_stream → workflow_streams; migrate to topic handles

- Directory and module path renamed to plural to match sdk-python
  `temporalio.contrib.workflow_streams` rename.
- Workflow-side: bind a typed topic handle in `@workflow.init` and call
  `topic.publish(value)` — the removed `WorkflowStream.publish` form is
  gone. Same change applied to the activity and external-publisher.
- Activity: `WorkflowStreamClient.from_activity()` →
  `from_within_activity()`.

Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>

Author: jssmith

README.md

@@ -79,7 +79,7 @@ Some examples require extra dependencies. See each sample's directory for specif
 * [patching](patching) - Alter workflows safely with `patch` and `deprecate_patch`.
 * [polling](polling) - Recommended implementation of an activity that needs to periodically poll an external resource waiting its successful completion.
 * [prometheus](prometheus) - Configure Prometheus metrics on clients/workers.
-* [workflow_stream](workflow_stream) - Workflow-hosted durable event stream via `temporalio.contrib.workflow_stream`. **Experimental — requires the [`contrib/pubsub` branch](https://github.com/temporalio/sdk-python/tree/contrib/pubsub) of sdk-python.**
+* [workflow_streams](workflow_streams) - Workflow-hosted durable event stream via `temporalio.contrib.workflow_streams`. **Experimental — requires the [`contrib/pubsub` branch](https://github.com/temporalio/sdk-python/tree/contrib/pubsub) of sdk-python.**
 * [pydantic_converter](pydantic_converter) - Data converter for using Pydantic models.
 * [schedules](schedules) - Demonstrates a Workflow Execution that occurs according to a schedule.
 * [sentry](sentry) - Report errors to Sentry.

workflow_stream/activities/payment_activity.py

@@ -1,15 +1,15 @@
 # Workflow Streams
 
 > **Experimental.** These samples target the
-> `temporalio.contrib.workflow_stream` module on the
+> `temporalio.contrib.workflow_streams` module on the
 > [`contrib/pubsub` branch of sdk-python][branch], which is not yet
 > released. To run them locally, install sdk-python from that branch
 > (e.g. `uv pip install -e <path-to-sdk-python>` after checking out the
 > branch).
 
 [branch]: https://github.com/temporalio/sdk-python/tree/contrib/pubsub
 
-`temporalio.contrib.workflow_stream` lets a workflow host a durable,
+`temporalio.contrib.workflow_streams` lets a workflow host a durable,
 offset-addressed event channel. The workflow holds an append-only log;
 external clients (activities, starters, BFFs) publish to topics via
 signals and subscribe via long-poll updates. This packages the
@@ -24,7 +24,7 @@ This directory has two scenarios sharing one Worker.
   `WorkflowStream` and publishes status events as it processes an order.
 * `activities/payment_activity.py` — an activity that publishes
   intermediate progress to the stream via
-  `WorkflowStreamClient.from_activity()`.
+  `WorkflowStreamClient.from_within_activity()`.
 * `run_publisher.py` — starts the workflow, subscribes to both topics,
   decodes each by `item.topic`, and prints events as they arrive.
 
@@ -73,16 +73,16 @@ This directory has two scenarios sharing one Worker.
 
 ```bash
 # Terminal 1: worker
-uv run workflow_stream/run_worker.py
+uv run workflow_streams/run_worker.py
 
 # Terminal 2: pick a scenario
-uv run workflow_stream/run_publisher.py
+uv run workflow_streams/run_publisher.py
 # or
-uv run workflow_stream/run_reconnecting_subscriber.py
+uv run workflow_streams/run_reconnecting_subscriber.py
 # or
-uv run workflow_stream/run_external_publisher.py
+uv run workflow_streams/run_external_publisher.py
 # or
-uv run workflow_stream/run_truncating_ticker.py
+uv run workflow_streams/run_truncating_ticker.py
 ```
 
 Expected output on the basic publisher side:

workflow_stream/README.md workflow_streams/README.mdworkflow_stream/README.md renamed to workflow_streams/README.md

@@ -0,0 +1,41 @@
+from __future__ import annotations
+
+import asyncio
+from datetime import timedelta
+
+from temporalio import activity
+from temporalio.contrib.workflow_streams import WorkflowStreamClient
+
+from workflow_streams.shared import TOPIC_PROGRESS, ProgressEvent
+
+
+@activity.defn
+async def charge_card(order_id: str) -> str:
+    """Pretend to charge a card, publishing progress to the parent workflow.
+
+    `WorkflowStreamClient.from_within_activity()` reads the parent
+    workflow id and the Temporal client from the activity context, so
+    this activity can push events back without any wiring.
+
+    Caveat: each call to ``from_within_activity()`` creates a fresh
+    client with a random ``publisher_id``, so dedup does not protect
+    against an activity retry republishing the same events. For
+    activities that must be exactly-once on the stream side, derive a
+    stable ``publisher_id`` from ``activity.info().activity_id`` (this
+    is invariant across attempts of the same scheduled activity). The
+    current ``WorkflowStreamClient`` API does not yet expose
+    ``publisher_id`` on its constructors; this sample accepts
+    at-most-once-per-attempt semantics.
+    """
+    client = WorkflowStreamClient.from_within_activity(
+        batch_interval=timedelta(milliseconds=200)
+    )
+    async with client:
+        progress = client.topic(TOPIC_PROGRESS, type=ProgressEvent)
+        progress.publish(ProgressEvent(message="charging card..."))
+        await asyncio.sleep(1.0)
+        progress.publish(
+            ProgressEvent(message="card charged"),
+            force_flush=True,
+        )
+    return f"charge-{order_id}"

workflow_stream/__init__.py workflow_streams/__init__.pyworkflow_stream/__init__.py renamed to workflow_streams/__init__.py

@@ -14,9 +14,9 @@
 ``HubWorkflow.close``, the workflow's run finishes, and the
 subscriber's iterator exits normally.
 
-Run the worker first (``uv run workflow_stream/run_worker.py``), then::
+Run the worker first (``uv run workflow_streams/run_worker.py``), then::
 
-    uv run workflow_stream/run_external_publisher.py
+    uv run workflow_streams/run_external_publisher.py
 """
 
 from __future__ import annotations
@@ -25,15 +25,15 @@
 import uuid
 
 from temporalio.client import Client
-from temporalio.contrib.workflow_stream import WorkflowStreamClient
+from temporalio.contrib.workflow_streams import WorkflowStreamClient
 
-from workflow_stream.shared import (
+from workflow_streams.shared import (
     TASK_QUEUE,
     TOPIC_NEWS,
     HubInput,
     NewsEvent,
 )
-from workflow_stream.workflows.hub_workflow import HubWorkflow
+from workflow_streams.workflows.hub_workflow import HubWorkflow
 
 
 HEADLINES = [
@@ -64,8 +64,9 @@ async def publish_news() -> None:
         # we know the events landed before the workflow shuts down.
         producer = WorkflowStreamClient.create(client, workflow_id)
         async with producer:
+            news = producer.topic(TOPIC_NEWS, type=NewsEvent)
             for headline in HEADLINES:
-                producer.publish(TOPIC_NEWS, NewsEvent(headline=headline))
+                news.publish(NewsEvent(headline=headline))
                 print(f"[publisher] sent: {headline}")
                 await asyncio.sleep(0.5)
             await producer.flush()

workflow_stream/activities/__init__.py workflow_streams/activities/__init__.pyworkflow_stream/activities/__init__.py renamed to workflow_streams/activities/__init__.py

@@ -5,9 +5,9 @@
 
 from temporalio.api.common.v1 import Payload
 from temporalio.client import Client
-from temporalio.contrib.workflow_stream import WorkflowStreamClient
+from temporalio.contrib.workflow_streams import WorkflowStreamClient
 
-from workflow_stream.shared import (
+from workflow_streams.shared import (
     TASK_QUEUE,
     TOPIC_PROGRESS,
     TOPIC_STATUS,
@@ -16,7 +16,7 @@
     StatusEvent,
     race_with_workflow,
 )
-from workflow_stream.workflows.order_workflow import OrderWorkflow
+from workflow_streams.workflows.order_workflow import OrderWorkflow
 
 
 async def main() -> None:

workflow_streams/activities/payment_activity.py

@@ -10,9 +10,9 @@
 the demo short. The same code shape works across actual process
 restarts because the resume offset is persisted to disk between phases.
 
-Run the worker first (``uv run workflow_stream/run_worker.py``), then::
+Run the worker first (``uv run workflow_streams/run_worker.py``), then::
 
-    uv run workflow_stream/run_reconnecting_subscriber.py
+    uv run workflow_streams/run_reconnecting_subscriber.py
 """
 
 from __future__ import annotations
@@ -23,15 +23,15 @@
 from pathlib import Path
 
 from temporalio.client import Client
-from temporalio.contrib.workflow_stream import WorkflowStreamClient
+from temporalio.contrib.workflow_streams import WorkflowStreamClient
 
-from workflow_stream.shared import (
+from workflow_streams.shared import (
     TASK_QUEUE,
     TOPIC_STATUS,
     PipelineInput,
     StageEvent,
 )
-from workflow_stream.workflows.pipeline_workflow import PipelineWorkflow
+from workflow_streams.workflows.pipeline_workflow import PipelineWorkflow
 
 # Number of events read in phase 1 before simulating a disconnect.
 # Picked small enough that the workflow is still running after.

workflow_stream/run_external_publisher.py …rkflow_streams/run_external_publisher.pyworkflow_stream/run_external_publisher.py renamed to workflow_streams/run_external_publisher.py workflow_stream/run_external_publisher.py renamed to workflow_streams/run_external_publisher.py

@@ -18,9 +18,9 @@
 the right trade — pair with set-semantic events where each event
 carries enough state to make missing the prior ones recoverable.
 
-Run the worker first (``uv run workflow_stream/run_worker.py``), then::
+Run the worker first (``uv run workflow_streams/run_worker.py``), then::
 
-    uv run workflow_stream/run_truncating_ticker.py
+    uv run workflow_streams/run_truncating_ticker.py
 """
 
 from __future__ import annotations
@@ -29,15 +29,15 @@
 import uuid
 
 from temporalio.client import Client
-from temporalio.contrib.workflow_stream import WorkflowStreamClient
+from temporalio.contrib.workflow_streams import WorkflowStreamClient
 
-from workflow_stream.shared import (
+from workflow_streams.shared import (
     TASK_QUEUE,
     TOPIC_TICK,
     TickerInput,
     TickEvent,
 )
-from workflow_stream.workflows.ticker_workflow import TickerWorkflow
+from workflow_streams.workflows.ticker_workflow import TickerWorkflow
 
 
 SLOW_SUBSCRIBER_DELAY_S = 1.5