Added the stream examples, one per path and one composed.

CHANGELOG.md

@@ -130,6 +130,8 @@ to include examples, links to docs, or any other relevant information.
   through `ExternalOutputStreamClient`. Workflow output is staged outside
   History and becomes readable only after its compact Workflow Task marker is
   committed.
+- Added `examples/streams`, one agent loop that runs unchanged on every
+  stream provider and on the Nexus front.
 
 ### Changed
 

examples/__init__.py

@@ -0,0 +1 @@
+"""Worked examples that run against a Temporal server."""

examples/streams/README.md

@@ -0,0 +1,57 @@
+# Streams, by path
+
+One provider, registered once on the client. Workers built from that client
+inherit it, and every context asks for its stream the same way. A topic is
+defined once, with the type its records carry, and every context refers to
+that definition, so no call names a type again.
+
+```python
+client = await Client.connect("localhost:7233", plugins=[provider])
+
+INPUTS = streams.topic("inputs", Token)
+DECISIONS = streams.topic("decisions", Decision)
+```
+
+| Path | Who | Call | Example file |
+|---|---|---|---|
+| A: the workflow publishes | workflow code | `workflow.stream_writer(PROGRESS).publish(Progress(...))`, then `.finish()` | `path_a_publish.py` |
+| A: a backend follows | any process with a client | `stream = client.get_stream_handle(workflow_id)`, then `stream.read(topic=PROGRESS, after=await stream.latest(topic=PROGRESS))` | `path_a_publish.py` |
+| B: an Activity produces | activity code | `await activity.stream_handle().producer(topic=INPUTS).append(Token(...))` | `path_b_produce.py` |
+| B: a backend produces | any process with a client | `client.get_stream_handle(workflow_id).producer(topic=NOTES, producer_id=..., attempt=...)` | `path_b_produce.py` |
+| B: a backend consumes | any process with a client | `client.get_stream_handle(workflow_id).read(topic=INPUTS)` | `path_b_produce.py` |
+| C: the workflow consumes | workflow code | `async for record in workflow.stream_reader(COMMANDS)` | `path_c_consume.py` |
+
+A plain string names a topic decided at runtime, with `result_type=` on the
+call; the examples never need one.
+
+`agent.py` and `run.py` compose all three paths in one agent, on every provider
+and behind the Nexus front. `_setup.py` is the one place a store is named.
+`june_scenarios/` maps every scenario in Roey's June design notes onto this
+surface, one file per scenario family, with a status on each.
+
+## Running
+
+Each example takes the provider's name and runs against a dev server:
+
+```sh
+python -m examples.streams.path_a_publish workflow_streams
+python -m examples.streams.path_a_publish native --address 127.0.0.1:7333
+python -m examples.streams.path_a_publish redis --redis redis://127.0.0.1:6379
+```
+
+Swap `path_a_publish` for `path_b_produce`, `path_c_consume` or `run`. The
+`native` provider needs a server built from the stream-carrying branch; the
+`redis` provider needs a Redis to point at. `run.py` also takes `nexus`, with
+an endpoint routed to the handler worker's task queue.
+
+## Why the workflow's verbs differ
+
+An Activity and a backend hold the same `StreamHandle`, with the same verbs,
+because both act on the store at once: a producer's records are visible as
+soon as the store accepts them, and a read follows the store live. Workflow
+code gets two verbs of its own because its semantics differ. `publish` is
+buffered and commits with the Workflow Task, so no reader can see a record
+from a task that failed, and a `stream_reader` is an observation the SDK
+records, so replay re-supplies the same records in the same order. That is
+why `publish` is a plain call and the reader is an async iterator, and why
+neither takes a client.

examples/streams/__init__.py

@@ -0,0 +1 @@
+"""One agent loop run on every stream provider."""

examples/streams/_setup.py

@@ -0,0 +1,60 @@
+"""Provider selection for the examples: the one place a store is named.
+
+Every example takes the provider's name on the command line, builds it here,
+and registers it once on the client. Nothing else in the examples names a
+store: workers built from the client inherit the provider, and each context
+asks for its stream through ``workflow.stream_reader`` or
+``workflow.stream_writer``, ``activity.stream_handle()`` and
+``client.get_stream_handle()``.
+"""
+
+from __future__ import annotations
+
+import argparse
+
+from temporalio.client import Client
+from temporalio.streams.providers import ProviderPlugin
+
+PROVIDERS = ("workflow_streams", "native", "redis")
+
+
+def parser(
+    description: str, providers: tuple[str, ...] = PROVIDERS
+) -> argparse.ArgumentParser:
+    """The flags every example shares."""
+    parser = argparse.ArgumentParser(description=description)
+    parser.add_argument("provider", choices=providers)
+    parser.add_argument("--address", default="localhost:7233")
+    parser.add_argument("--redis", default="redis://127.0.0.1:6379")
+    return parser
+
+
+def make_provider(name: str, args: argparse.Namespace) -> ProviderPlugin:
+    """The whole difference between the stores: one constructor call."""
+    if name == "workflow_streams":
+        from temporalio.streams.providers.workflow_streams import (
+            WorkflowStreamsProvider,
+        )
+
+        return WorkflowStreamsProvider()
+    if name == "redis":
+        from temporalio.streams.providers.redis import RedisStreams
+
+        return RedisStreams(url=args.redis)
+    if name == "native":
+        from temporalio.streams.providers.native import NativeStreams
+
+        return NativeStreams()
+    if name == "memory":
+        # Only for examples that keep a warm cache: this provider is not
+        # replay-safe, which is why PROVIDERS leaves it out.
+        from temporalio.streams.providers.memory import MemoryStreams
+
+        return MemoryStreams()
+    raise SystemExit(f"unknown provider {name}")
+
+
+async def connect(args: argparse.Namespace) -> tuple[Client, ProviderPlugin]:
+    """A client with the provider registered on it, and the provider to close later."""
+    provider = make_provider(args.provider, args)
+    return await Client.connect(args.address, plugins=[provider]), provider

examples/streams/agent.py

@@ -0,0 +1,125 @@
+"""The workflow and activities. Identical on every provider.
+
+Nothing here names a store, a transport, or an option. The two topics are
+defined once, with the types their records carry, and the workflow, the
+Activity and the backend in ``run.py`` all refer to them. The loop reads its
+``inputs`` topic, decides, publishes the decision, and runs an ordinary
+activity in the same workflow task, which is the shape the design doc calls
+Paths A, B and C together. The Activity that streams model output asks its
+context for its own workflow's stream, the way workflow code asks its
+runtime, so the file is the same whichever provider the process registered.
+"""
+
+from __future__ import annotations
+
+import asyncio
+from dataclasses import dataclass
+from datetime import timedelta
+
+from temporalio import activity, streams, workflow
+from temporalio.common import RetryPolicy
+from temporalio.streams import RecordKind
+
+
+@dataclass
+class Token:
+    """One piece of model output."""
+
+    n: int
+
+
+@dataclass
+class Decision:
+    """What the workflow decided about a token, or which attempt it retracted."""
+
+    echo: int | None = None
+    retracting_attempt: int | None = None
+
+
+INPUTS = streams.topic("inputs", Token)
+DECISIONS = streams.topic("decisions", Decision)
+
+
+@activity.defn
+async def generate(count: int) -> None:
+    """Stream model output onto this workflow's ``inputs`` topic.
+
+    No workflow id and no run id: the handle is this Activity's own
+    workflow, pinned to its run. The producer carries the Activity's own id
+    and attempt, so a retry deduplicates and a new attempt is reported to
+    readers as a supersession.
+    """
+    model = activity.stream_handle().producer(topic=INPUTS)
+    for n in range(count):
+        await model.append(Token(n))
+    await model.finish()
+
+
+@activity.defn
+async def record_decision(decision: Decision) -> str:
+    """An ordinary activity, run from the same task that read and published."""
+    return f"recorded {decision.echo}"
+
+
+@workflow.defn
+class Agent:
+    """Reads ``inputs``, publishes a decision each time, ends on FINISH."""
+
+    @workflow.run
+    async def run(self, count: int) -> int:
+        """Decide on at most ``count`` inputs, then return how many landed."""
+        decisions = workflow.stream_writer(DECISIONS)
+
+        generating = workflow.start_activity(
+            generate,
+            count,
+            start_to_close_timeout=timedelta(minutes=1),
+            # Bounded, so a generator that cannot finish gives up instead of
+            # retrying forever while every attempt streams from the start.
+            retry_policy=RetryPolicy(maximum_attempts=3),
+        )
+        consuming = asyncio.create_task(self._consume(count, decisions))
+
+        # Raced rather than awaited in turn: an attempt that fails writes no
+        # FINISH, so a generator that exhausts its attempts leaves the reader
+        # waiting forever. Its failure ends the run with its cause instead.
+        done, _ = await workflow.wait(
+            [consuming, generating], return_when=asyncio.FIRST_COMPLETED
+        )
+        if generating in done and consuming not in done:
+            try:
+                await generating
+            except BaseException:
+                consuming.cancel()
+                raise
+        seen = await consuming
+        await generating
+
+        decisions.finish()
+        return seen
+
+    async def _consume(
+        self, count: int, decisions: workflow.StreamWriter[Decision]
+    ) -> int:
+        seen = 0
+        async for record in workflow.stream_reader(INPUTS):
+            if record.kind is RecordKind.FINISH:
+                break
+            if record.kind is RecordKind.SUPERSEDED:
+                assert record.supersession is not None
+                decisions.publish(
+                    Decision(retracting_attempt=record.supersession.previous_attempt)
+                )
+                continue
+            assert record.value is not None
+            seen += 1
+            decision = Decision(echo=record.value.n)
+            decisions.publish(decision)
+            await workflow.execute_activity(
+                record_decision,
+                decision,
+                start_to_close_timeout=timedelta(minutes=1),
+            )
+            if seen >= count:
+                break
+        return seen

examples/streams/june_scenarios/README.md

@@ -0,0 +1,53 @@
+# Roey's June scenarios, on the shipped surface
+
+Every scenario in Roey's Notion page "Streaming Design Discussion Prep Notes"
+(June 2, under "Streaming Links") mapped onto `temporalio.streams` as it
+ships on this branch. Each file opens with his scenario heading, a status,
+and one sentence why. Where his sketch uses a call shape we do not have, the
+docstring shows his shape in one line and the code uses ours. Nothing here
+reaches into private SDK code or adds a feature.
+
+| Roey's scenario | File | Status | Note |
+|---|---|---|---|
+| Client starts and consume stream: primary and named | `s1_client_consumes.py` | implemented | Default topic with no name, a typed topic, `last=N`, `after=END`, `BEGINNING` on a moved floor (memory only) |
+| Client starts and consume stream: standalone alt 1, 2, 3 | `s2_standalone_streams.py` | alts 1, 2 and 3 implemented | Alt 1 as a read that parks until `create_stream` and an append land (native; memory and Redis answer `StreamNotFoundError`); alt 2 as `client.create_stream`, a policy floor, `close()` and `StreamClosedError`; alt 3 as a stream created first and passed into the workflow start as a `StreamRef`, opened in the activity with `activity.stream_handle(ref)`. A start that commits the stream with the workflow remains the design question. Workflow Streams declines |
+| Workflow as Producer: as named handle | `s3_workflow_producer.py` | implemented | His turn loop with continue-as-new; the client follows the chain live |
+| Workflow as Producer: as return type | `s4_workflow_as_generator.py` | emulated | Default-topic publishes plus `FINISH`, result from the workflow; the generator signature is sugar not built |
+| Activity as Producer: as named handle | `s5_activity_producers.py` | implemented | Workflow topic (Path B), `scope="activity"`, standalone activity; all three on native, memory and Redis, the last two skipped on Workflow Streams |
+| Activity as Producer: as return type | `s6_activity_as_generator.py` | emulated | Appends plus a heartbeat checkpoint; the retry resumes and readers see `SUPERSEDED` |
+| Workflow as Consumer | `s7_workflow_consumer.py` | implemented; foreign stream unsupported | Own inbound topic across continue-as-new, handing over per batch with one producer per batch and carrying a checkpoint; runs on native, Workflow Streams and Redis, memory skips by design; reading a foreign stream from a workflow is rule 5 |
+| Client as Consumer over Standalone Nexus | `s8_nexus_consumers.py` (a) | implemented | Activity reads through the `NexusStreams` front and resumes from a heartbeat cursor |
+| Nexus operation handler | `s8_nexus_consumers.py` (b) | implemented | The operation returns `temporalio.streams.StreamRef`, taken from the producing workflow's handle, and the client opens it with `get_stream_handle(ref)` on a client whose provider is the front; a stream type of its own in the operation IDL is the nexgen follow-on |
+| Workflow as Consumer over Nexus | `s8_nexus_consumers.py` docstring | unsupported by design | A workflow's reads ride its Workflow Task and never cross Nexus |
+
+## Running
+
+Each file runs on its own and takes the provider's name, the same way the
+examples one directory up do. `run.py` runs them all in order:
+
+```sh
+python -m examples.streams.june_scenarios.run native --address 127.0.0.1:7433 --http http://127.0.0.1:7343
+python -m examples.streams.june_scenarios.run workflow_streams --address 127.0.0.1:7433 --http http://127.0.0.1:7343
+python -m examples.streams.june_scenarios.run memory --address 127.0.0.1:7433 --http http://127.0.0.1:7343
+python -m examples.streams.june_scenarios.run redis --address 127.0.0.1:7433 --http http://127.0.0.1:7343 --redis redis://127.0.0.1:6379
+python -m examples.streams.june_scenarios.s2_standalone_streams native --address 127.0.0.1:7433
+```
+
+`native` needs a server built from the stream-carrying branch, and `s2`
+alt 1's read that parks until the stream is created needs one built from
+its current head. `s5` (b) and (c) need a server with standalone activities
+and activity-owned streams, and `s8` needs the server's Nexus HTTP ingress
+(`--http`, default `http://127.0.0.1:7243`, `7343` on the server above);
+the stream-carrying server has all of them, so the commands above point
+every provider at it. `s8` creates and deletes its own Nexus endpoint.
+`memory` is offered here, not in the parent examples, because it is not
+replay-safe; these scenarios keep a warm cache. `memory`, `native` and
+`redis` hold standalone streams, so `s2` runs on all three; `redis` runs
+with `--redis` naming a local Redis. Every scenario ran green on all four
+providers against that server, apart from the refusals below.
+
+A scenario a provider cannot serve says so in its output and moves on:
+`s2` on `workflow_streams`, which keeps a stream inside a workflow's log,
+`s5` (b) and (c) on `workflow_streams`, `s1` (d) on anything but `memory`,
+and `s7` on `memory`, which keeps one topic across a chain rather than one
+per run.

examples/streams/june_scenarios/__init__.py

@@ -0,0 +1 @@
+"""Roey's June streaming scenarios, each mapped onto the shipped streams surface."""

examples/streams/june_scenarios/_common.py

@@ -0,0 +1,38 @@
+"""What every scenario shares: its flags, its ids and one way to print.
+
+The store is still named in one place, ``examples.streams._setup``. The
+scenarios add the memory provider to the choices because two of them need an
+activity-owned stream or truncation, and memory is the one in-process store
+that has both.
+"""
+
+from __future__ import annotations
+
+import argparse
+import uuid
+
+from examples.streams import _setup
+
+PROVIDERS = (*_setup.PROVIDERS, "memory")
+
+
+def parser(description: str) -> argparse.ArgumentParser:
+    """The example flags, plus the memory provider and the Nexus ingress."""
+    parser = _setup.parser(description, PROVIDERS)
+    parser.add_argument(
+        "--http",
+        default="http://127.0.0.1:7243",
+        help="the server's Nexus HTTP ingress, for s8",
+    )
+    return parser
+
+
+def ids(prefix: str) -> tuple[str, str]:
+    """A fresh workflow id and its own task queue, so reruns never collide."""
+    workflow_id = f"{prefix}-{uuid.uuid4().hex[:8]}"
+    return workflow_id, f"tq-{workflow_id}"
+
+
+def banner(title: str, provider: str) -> None:
+    """Head each scenario's output, so a run of all of them reads in sections."""
+    print(f"\n== {title} [{provider}]")