workflow_streams: deliver terminal events + fix run_publisher subscribe shape

End-to-end runs of the four workflow_streams scenarios surfaced two
sample-side issues, both fixed here.

run_publisher's consumer asserted ``isinstance(item.data, Payload)`` and
called ``payload_converter.from_payload(item.data, T)``. The contrib's
``subscribe()`` defaults to converter-decoded data, not raw payloads,
so this assertion fired on the first run. Switch to
``result_type=RawValue`` (the documented escape hatch for heterogeneous
topics) and read ``item.data.payload``.

Items published in the same workflow task that returns from
``@workflow.run`` were not delivered to subscribers — the in-memory
log dies with the workflow and the next subscriber poll lands on a
completed workflow. Fix: each scenario now uses an in-band terminator
that subscribers break on, and each workflow holds the run open with
``await workflow.sleep(timedelta(milliseconds=500))`` so that final
publish is fetched before the workflow exits:

- OrderWorkflow / PipelineWorkflow: the workflow's own
  ``StatusEvent(kind="complete")`` / ``StageEvent(stage="complete")``
  is the terminator (consumers already broke on it).
- HubWorkflow: the *publisher* in run_external_publisher emits a
  sentinel ``NewsEvent(headline="__done__")`` immediately before
  signaling close; the consumer breaks on the sentinel.
- TickerWorkflow: the final tick (n == count - 1) is the terminator;
  ``keep_last`` guarantees that offset survives the last truncation,
  so even slow consumers reach it.

Because subscribers stop polling on the terminator, by the time
``workflow.run`` returns there are no in-flight poll handlers — no
``UnfinishedUpdateHandlersWarning`` from the SDK and no need for
``detach_pollers()`` / ``wait_condition(all_handlers_finished)`` in
the workflow exit path.

Two consecutive end-to-end runs of all four scenarios pass cleanly
against ``temporal server start-dev --headless``.

Author: jssmith

workflow_streams/run_external_publisher.py

@@ -44,6 +44,13 @@
     "regulator opens probe",
 ]
 
+# In-band terminator the publisher emits before signaling close. The
+# subscriber recognizes this value and stops polling — without an
+# explicit terminator the consumer would have to rely on the workflow
+# returning to break the iterator, which means racing the last item
+# delivery against workflow completion.
+DONE_HEADLINE = "__done__"
+
 
 async def main() -> None:
     client = await Client.connect("localhost:7233")
@@ -69,9 +76,10 @@ async def publish_news() -> None:
                 news.publish(NewsEvent(headline=headline))
                 print(f"[publisher] sent: {headline}")
                 await asyncio.sleep(0.5)
+            news.publish(NewsEvent(headline=DONE_HEADLINE), force_flush=True)
             await producer.flush()
-        # Tell the hub it can stop. The workflow's run() returns, and
-        # any in-flight subscribers see their async-for loop exit.
+        # Tell the hub it can stop. The subscriber has already broken
+        # out of its async-for loop on the sentinel above.
         await handle.signal(HubWorkflow.close)
         print("[publisher] signaled close")
 
@@ -80,6 +88,8 @@ async def consume_news() -> None:
         async for item in consumer.subscribe(
             [TOPIC_NEWS], result_type=NewsEvent
         ):
+            if item.data.headline == DONE_HEADLINE:
+                return
             print(f"[subscriber] offset={item.offset}: {item.data.headline}")
 
     await asyncio.gather(publish_news(), consume_news())

workflow_streams/run_publisher.py

@@ -3,8 +3,8 @@
 import asyncio
 import uuid
 
-from temporalio.api.common.v1 import Payload
 from temporalio.client import Client
+from temporalio.common import RawValue
 from temporalio.contrib.workflow_streams import WorkflowStreamClient
 
 from workflow_streams.shared import (
@@ -35,17 +35,19 @@ async def main() -> None:
 
     async def consume() -> None:
         # Single iterator over both topics — avoids a cancellation race
-        # between two concurrent subscribers. result_type is left unset
-        # so we can dispatch heterogeneous events on item.topic.
-        async for item in stream.subscribe([TOPIC_STATUS, TOPIC_PROGRESS]):
-            assert isinstance(item.data, Payload)
+        # between two concurrent subscribers. result_type=RawValue
+        # delivers the underlying Payload so we can dispatch
+        # heterogeneous events on item.topic.
+        async for item in stream.subscribe(
+            [TOPIC_STATUS, TOPIC_PROGRESS], result_type=RawValue
+        ):
             if item.topic == TOPIC_STATUS:
-                evt = converter.from_payload(item.data, StatusEvent)
+                evt = converter.from_payload(item.data.payload, StatusEvent)
                 print(f"[status] {evt.kind}: order={evt.order_id}")
                 if evt.kind == "complete":
                     return
             elif item.topic == TOPIC_PROGRESS:
-                progress = converter.from_payload(item.data, ProgressEvent)
+                progress = converter.from_payload(item.data.payload, ProgressEvent)
                 print(f"[progress] {progress.message}")
 
     result = await race_with_workflow(consume(), handle)

workflow_streams/run_truncating_ticker.py

@@ -41,6 +41,7 @@
 
 
 SLOW_SUBSCRIBER_DELAY_S = 1.5
+TICKER_COUNT = 20
 
 
 async def main() -> None:
@@ -50,7 +51,7 @@ async def main() -> None:
     handle = await client.start_workflow(
         TickerWorkflow.run,
         TickerInput(
-            count=20,
+            count=TICKER_COUNT,
             keep_last=3,
             truncate_every=5,
             interval_ms=400,
@@ -60,19 +61,24 @@ async def main() -> None:
     )
 
     stream = WorkflowStreamClient.create(client, workflow_id)
+    last_n = TICKER_COUNT - 1
 
+    # Both subscribers break on the final tick (n == last_n). ``keep_last``
+    # ensures that offset survives the last truncation so even the slow
+    # consumer reaches it.
     async def fast_subscriber() -> None:
         async for item in stream.subscribe([TOPIC_TICK], result_type=TickEvent):
             print(f"[fast]  offset={item.offset:3d}  n={item.data.n}")
+            if item.data.n == last_n:
+                return
 
     async def slow_subscriber() -> None:
         async for item in stream.subscribe([TOPIC_TICK], result_type=TickEvent):
             print(f"[SLOW]  offset={item.offset:3d}  n={item.data.n}")
+            if item.data.n == last_n:
+                return
             await asyncio.sleep(SLOW_SUBSCRIBER_DELAY_S)
 
-    # Both iterators exit normally when the workflow completes. No
-    # terminal sentinel is needed — see the doc's "When the Workflow
-    # run completes" note.
     await asyncio.gather(fast_subscriber(), slow_subscriber())
 
     result = await handle.result()

workflow_streams/workflows/hub_workflow.py

@@ -1,5 +1,7 @@
 from __future__ import annotations
 
+from datetime import timedelta
+
 from temporalio import workflow
 from temporalio.contrib.workflow_streams import WorkflowStream
 
@@ -26,6 +28,11 @@ def __init__(self, input: HubInput) -> None:
     @workflow.run
     async def run(self, input: HubInput) -> str:
         await workflow.wait_condition(lambda: self._closed)
+        # The publisher publishes its own terminator into the stream
+        # before signaling close (see run_external_publisher.py).
+        # Hold the run open briefly so subscribers' final poll
+        # delivers any items still in the log.
+        await workflow.sleep(timedelta(milliseconds=500))
         return f"hub {input.hub_id} closed"
 
     @workflow.signal

workflow_streams/workflows/order_workflow.py

@@ -50,4 +50,9 @@ async def run(self, input: OrderInput) -> str:
         self.status.publish(StatusEvent(kind="shipped", order_id=input.order_id))
         self.progress.publish(ProgressEvent(message=f"charge id: {charge_id}"))
         self.status.publish(StatusEvent(kind="complete", order_id=input.order_id))
+        # The "complete" status event above is the in-band terminator
+        # subscribers break on (see run_publisher.py). Hold the run
+        # open briefly so subscribers' next poll delivers it before
+        # this task returns and the in-memory log is gone.
+        await workflow.sleep(timedelta(milliseconds=500))
         return charge_id

workflow_streams/workflows/pipeline_workflow.py

@@ -41,4 +41,8 @@ async def run(self, input: PipelineInput) -> str:
             self.status.publish(StageEvent(stage=stage))
             if stage != "complete":
                 await workflow.sleep(timedelta(seconds=2))
+        # The "complete" stage above is the in-band terminator
+        # subscribers break on. Hold the run open briefly so the final
+        # poll delivers it.
+        await workflow.sleep(timedelta(milliseconds=500))
         return f"pipeline {input.pipeline_id} done"

workflow_streams/workflows/ticker_workflow.py

@@ -57,4 +57,10 @@ async def run(self, input: TickerInput) -> str:
                 # Drop everything except the last `keep_last` entries.
                 truncate_to = self._published - input.keep_last
                 self.stream.truncate(truncate_to)
+        # The final tick (n == count - 1) is the in-band terminator
+        # subscribers break on. ``keep_last`` guarantees that final
+        # offset survives the last truncation so even slow consumers
+        # eventually see it. Hold the run open briefly so the final
+        # poll delivers it.
+        await workflow.sleep(timedelta(milliseconds=500))
         return f"ticker emitted {self._published} events"