[moetemp]

What changed?

This PR adds the Streams API to the Python SDK. It's experimental.

A Stream is an ordered log of Records, for example the tokens of a model call. A Workflow, an Activity or a client writes it, and any of them reads it.

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()

The public API is the code that nexgen generates from the Stream WIT, which is in #2128. This PR is the hand-written part.

• Stores. StreamConfig gives a client its Stores, in the shape of ExternalStorage, and a Worker uses the config of its client. A call goes to the Store in its ref, else to the only Store, else to the one that the selector returns. RedisStreamStore and MemoryStreamStore run Core's temporalio-streams crate through the bridge. You can also write a Store in Python.
• Workflow code can create, get, append to, close and describe a Stream. It has no read yet. A create, an append and a describe each run as a Local Activity, so a replay sends nothing to the Store. A close runs as a regular Activity, so it reaches the Store only after its Workflow Task completes. For that reason, a Worker with Workflows and a Stream config polls for Activity Tasks.
• Fencing. A write from a Local Activity uses its Workflow Task as its epoch (feat: add the dispatching Workflow Task to activity.Info for a Local Activity #2127), so the Store refuses the late writes of a failed Workflow Task. append also takes an optional sequence, so a retried batch is written once.
• Batching. temporalio.contrib.streams.BufferedAppender collects appends into one write.

It depends on three things that have no release yet.

• Core. The submodule points to the branch moe/streams-crate-for-sdk-python, which is the Core commit of main plus feat: add the Stream protos and WIT to the vendored API sdk-rust#1733 and feat: add temporalio-streams, the Stream runtime that SDK bridges link sdk-rust#1799.
• 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 nexgen's main, so poe gen-protos gets a second generator from NEX_GEN_STREAM_BIN.

Please note that the top-level README.md has no section on Streams yet.

Why?

An agent application shows partial output, such as the tokens of a model call, while the work still runs. Workflow History isn't the place for that data, so a Stream keeps its Records in a Store outside History.

The API is generated and the Stores run in Core, so the other SDKs can get the same API and the same Stores.

How did you test it?

The tests are in tests/nexus/test_stream_store*.py, tests/contrib/streams/ and tests/contrib/redis/. Most of them run on the local dev server. The Redis tests use a real Redis in CI on Linux and macOS, and they skip on Windows.

Link to a test plan if any -

• Unit Tests
• Staging
• End to End Tests