[moetemp]

This PR adds the Streams API to the Python SDK. A Stream is an ordered log of Records that a Workflow, an Activity or a client writes, and that any of them reads, for example the tokens of a model call. It is experimental.

The public API is the code that nexgen generates from the Stream WIT, which is in #2128. This PR is the hand-written part. It adds the Stream Stores, the config that gives them to a client, and the hooks that send each generated call to a Store.

from temporalio.client import Client, StreamConfig
from temporalio.contrib.redis.streams import RedisStreamStore

store = RedisStreamStore("redis://localhost:6379")
client = await Client.connect("localhost:7233", stream_config=StreamConfig(stores=[store]))

stream = await client.create_stream(id="chat-43", body_type=str)
await stream.append("Hello", " world")
async for record in stream.iter_read_items():
    print(record.body)
await stream.close()

Where to look

Part	Files	What it has
Stores	temporalio/client/_stream_store.py, temporalio/contrib/redis/streams.py, temporalio/testing/_stream_store.py	StreamStore, a public interface with the five operations and two Worker hooks. StreamConfig, with the Stores and the selector. RedisStreamStore and MemoryStreamStore.
Client hook	temporalio/client/_impl.py, temporalio/client/_client.py	Sends a Stream operation of the generated client to the Store of its ref.
Workflow hook	temporalio/worker/_stream_store.py, _workflow_instance.py, _worker.py, _replayer.py	Answers a Stream call of Workflow code through the Store. The Worker registers the Activities that each Store brings.
Support of the generated code	temporalio/nexus/system/_support.py, temporalio/nexus/system/__init__.py	The Producer of a call, the request id of a create, the poll helpers of a read, and the conversion between a model and its proto message.
Appender	temporalio/contrib/streams/	BufferedAppender, which batches appends. It is a library on top of the generated API, so it is in contrib.
Bridge	temporalio/bridge/src/streams.rs, temporalio/bridge/streams.py, temporalio/bridge/Cargo.toml	One call into Core's temporalio-streams runtime, behind the streams and streams-redis features.
Errors	temporalio/exceptions.py	StreamError with a reason, and StreamConfigError.
Generation	scripts/gen_nexus_system_api.py, scripts/gen_payload_visitor.py	The Stream package and its payload visitor in poe gen-protos.

How it behaves

• Config. A client takes stream_config=StreamConfig(stores=[...], store_selector=...), in the shape of ExternalStorage. A Worker takes the config from its client and serves every Store in it.
• Routing. A Stream operation goes to the Store that its ref names, else to the only Store, else to the one that the selector returns. A config can hold several Stores with no selector, and a call that names no Store then raises StreamConfigError. With no config, a client or Activity call raises StreamConfigError, since this release has no Streams on the server. In Workflow code that case fails the Workflow with a non-retryable ApplicationError of type StreamConfigError, so a Worker with a wrong setup is visible.
• Workflow code can create, get, append to, close and describe a Stream. The Store answers a create, an append and a describe through a Local Activity of its own, so a replay sends nothing to the Store. A close runs the same Activity as a regular Activity, so it reaches the Store only after its Workflow Task completes. The Workflow Stream has no read yet.
• Fencing. A write from a Local Activity uses its Workflow Task as its epoch, so a Store fences the writes of a failed Workflow Task. append also takes an optional sequence, and a batch that has one is written once. A Workflow append sends one, so a retry of its Local Activity in the same Workflow Task writes once.
• Workers. A Worker with Workflows and a Stream config polls for Activity Tasks, since a Workflow's close is a regular Activity.
• Batching. temporalio.contrib.streams.BufferedAppender collects the appends of up to max_delay into one batch, with one write in flight. In Workflow code it uses no timer, and the appends of one activation go as one write.
• Stores. RedisStreamStore keeps Streams in Redis, and MemoryStreamStore keeps them in memory for tests. Both run Core's temporalio-streams crate through the bridge. A person can also write a Store in Python.
• Errors. A failed Stream call raises StreamError with a reason. A wrong setup of the Stores raises StreamConfigError.

What it depends on

• The stack. feat: add the dispatching Workflow Task to activity.Info for a Local Activity #2127 adds the Workflow Task info of a Local Activity, chore: add the generated code of the Streams API #2128 adds the generated code on top of it, and this PR sits on chore: add the generated code of the Streams API #2128.
• Core. The submodule points at the branch moe/streams-crate-for-sdk-python of temporalio/sdk-rust. It has the commits of feat: add the Stream protos and WIT to the vendored API sdk-rust#1733 (the Stream protos and WIT) and feat: add temporalio-streams, the Stream runtime that SDK bridges link sdk-rust#1799 (the temporalio-streams crate) on top of the Core commit that main uses. So this PR changes nothing else in Core, and a Core update replaces the branch once those two merge.
• The Redis Store is a git dependency on feat: count a Stream's Retention from its close temporal-community/sdk-streams#8.
• nexgen. The Stream package needs the main of nexgen, which no release has yet. So poe gen-protos takes a second generator from NEX_GEN_STREAM_BIN, and the other packages stay on the released pin.

Please note that the README has no section on Streams yet.