Streams API: combined branch for review by moetemp · Pull Request #381 · moetemp/sdk-python

moetemp · 2026-11-09T17:23:11Z

This PR restarts the Streams API in sdk-python on the code that nexgen generates.

The public API is now the generated code. client.create_stream and client.get_stream are generated methods, and temporalio.workflow exports the generated Workflow functions. The generated files are in #382, the base of this PR, and #383 below it adds the Workflow Task info of a Local Activity. So this diff is only the hand-written Stream code. The earlier branch is kept at backup/pr-381-streams-combined-1112.

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. With a config, a Stream operation goes to the Store that its ref names, else to the only Store, else to the one that the selector returns. With no config, a Stream call raises StreamConfigError, since this release has no Streams on the server.
Workflow code can create, get and describe a Stream. The Store answers them through a Local Activity of its own, so a replay sends nothing to the Store. The Workflow Stream has no append, close or read in this release. To write, a Workflow calls its own Local Activity, which appends through activity.client().
Fencing. A write from a Local Activity uses the 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.
Batching. BufferedAppender collects the appends of up to max_delay into one batch, with one write in flight.
Stores. StreamStore is a public interface with the five operations as methods, so a person can write a Store. It also owns its Worker side, with worker_activities() and call_from_workflow(). RedisStreamStore keeps Streams in Redis, and MemoryStreamStore keeps them in memory for tests. Both run Core's temporalio-streams crate through the bridge.
Errors. A failed Stream call raises StreamError with a reason. A wrong setup of the Stores raises StreamConfigError.

The Stream WIT and protos come from temporalio/api#895, through the sdk-core submodule at moetemp/sdk-rust#228 1addba05. The Stream package needs the main of nexgen (9653a0f5), 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.

sdk-python takes its API files from Core's vendored copy. That commit adds the Stream protos and WIT from the api repo until the api Stream change merges.

The Stream package needs a newer nexgen than the existing packages, so it has its own pin until the api WITs and a nexgen release agree. The payload visitor covers every generated system Nexus package, and mypy skips the Stream package like the other generated one.

Written by `poe gen-protos` with NEX_GEN_BIN at nexgen 0.2.8 and NEX_GEN_STREAM_BIN at a build of nexgen commit bf3c49c9. Nothing in this commit is edited by hand.

The generated code calls these by name for request ids, Producers and read paging. The system Nexus serialization lookup also knows the Stream operations.

Client inherits the generated extension, so its Stream calls are the generated ones. A Store answers them in place of the server, with the conversion and the codec of the system Nexus path.

Workflow code can't do I/O, so a Worker whose client has a Stream Store calls it from an internal Local Activity. The Replayer takes the Stores too, so a replay makes the same commands.

Core sends the Workflow Task that dispatched a Local Activity. A write from a Local Activity needs it for its Producer epoch.

On a Store, a Workflow appends as a Local Activity Producer, and the Store refuses the Workflow Producer. Its epoch is the Workflow Task that ran the Local Activity, so the Store fences the write of a failed task.

The bridge links the Stream runtime from Core's new `temporalio-streams` crate. The Core commits add no proto that the SDK generates from.

A Store on the bridge needs the crate's runtime and its memory Store. The feature is off by default, and the build tasks and wheels turn it on.

Client, Activity and Workflow code can catch one class and match on its reason. The details carry the reason, so Workflow code rebuilds the same error from a Local Activity's failure.

Tests now run on the same Stream runtime and rules as every Store on the bridge. They check Records and Producers with the generated read, so the fake Store is gone.

Each run of a chain sends the same strict create. With one request id for the chain, the Store answers a later run's create as created and not as ALREADY_EXISTS.

tconley1428 · 2026-11-13T19:15:14Z

        impl: ActivityInboundInterceptor = _ActivityInboundImpl(self, running_activity)
-        for interceptor in reversed(list(self._interceptors)):
-            impl = interceptor.intercept_activity(impl)
+        # Users see a Stream call through the Workflow interceptors, so this internal


Don't understand that

A Workflow's Stream call already goes through the user's Workflow interceptors, as one system Nexus operation. This Activity is how the Worker runs that call. Without the skip, the user's Activity interceptors would also see an Activity that's not scheduled by user code. A tracing interceptor would show it as an orphan span, for example. I rewrote the comment to say that. Wdyt?

tconley1428 · 2026-11-13T19:17:07Z

+        activities = list(config.get("activities") or [])
+        stream_stores = client_config["stream_stores"]
+        if stream_stores and config.get("workflows"):
+            activities.append(


This is a bit sketchy. I follow the point though

Probably we'd be better off with a redis plugin that does the activity + stream store.

I see the appeal. My concern with a plugin is that a user would mention the Stores twice, once in stream_config and once in the plugin. Or the Stores move out of the client option, which the API review chose to match external storage. So the Worker still takes its Stores from its client, and the user sets them up in one place. To make it less "sketchy", the Activity and the Workflow routing sit in one module next to the Store as of 1c836c8.

tconley1428 · 2026-11-13T19:18:05Z

+        # Workflow code builds the request, so the Local Activity does only the I/O.
+        payload_converter, _ = self._system_nexus_converters(input)
+        [request] = payload_converter.to_payloads([input.input])
+        activity = self._outbound_schedule_activity(


I suspect this should probably live inside the stream store itself.

Most of the code in this function, likely

Moved it in 1c836c8.

tconley1428 · 2026-11-13T19:18:41Z

-    ) -> _NexusOperationHandle[OutputT]:
+    ) -> temporalio.workflow.NexusOperationHandle[OutputT]:
+        options = self._stream_store_call_options
+        if options is not None and input.service == _STREAM_SERVICE.name:


Don't know why we would check options is not none.

It meant "this Worker's client has a Store", and it wasn't obvious. The check is on the Worker's Stream config now. With none, a Stream call goes to the server like every system Nexus operation.

tconley1428 · 2026-11-13T19:23:46Z

 select = "*musllinux*"
-environment = { PATH = "$PATH:$HOME/.cargo/bin", CARGO_NET_GIT_FETCH_WITH_CLI = "true", RUSTFLAGS = "-C target-feature=-crt-static" }
+environment = { PATH = "$PATH:$HOME/.cargo/bin", CARGO_NET_GIT_FETCH_WITH_CLI = "true", RUSTFLAGS = "-C target-feature=-crt-static", MATURIN_PEP517_ARGS = "--features streams-redis" }



Seems like quite a bit of shenanigans involved here, I'll ignore the pyproject until we're further along in core.

Makes sense. Most of it is there so the dev build and the wheels link the Stream runtime while it's behind a cargo feature. The rest keeps the linters out of the Core submodule.

tconley1428 · 2026-11-13T19:25:08Z

        if: ${{ matrix.clippyLinter }}
      - run: poe build-develop
      - run: poe lint
+      - uses: ./.github/actions/install-redis


Not 100% on installing (and starting, which is unclear) a redis server for all tests, but maybe

The step installs Redis and starts one server on a port. Only the Redis Store tests use it, and they skip when its URL isn't set, as on Windows. I mentioned the step in ed82555 so that's clear at the call site. I can move those tests to their own job if you'd rather keep Redis out of the main one.

The Activity, the Local Activity call and its handle were spread over generic Worker files. One module now holds all that a Worker does for the Stream Store of its client.

A Store is the Rust runtime behind the bridge, so no user writes one in Python. Its call takes the bridge's operation name and bytes, which is not an API for users.

A reader asked why a list takes one Store. A named CI step shows why Redis is installed.

The agreed API holds the Stores and the rule that picks one in a single config, as `ExternalStorage` holds drivers and a selector. A ref names its Store, so a call on a known Stream reaches the right Store with several of them.

A test that starts its own server needs a client with Stores, as a test with `Client.connect` has. `start_local` and `start_time_skipping` give the config to the client that they build.

A Workflow's Stream call runs in the SDK's Local Activity, so the Store can be picked there by the client's rules. History keeps the ref with the Store's name, so a replay calls no selector.

A Workflow's create picks its Store only in the Local Activity, so with several Stores it could not use the options of any one Store. The config now holds one set of options for every Store.

Decoding the encoded request to find its ref decoded every Record of an append once more. The caller has the request object before it encodes it, and the Local Activity gets the ref as two strings.

An Activity that streams model tokens appends a few characters many times a second, and one Store call for each is too many. The appender writes them in batches of at most 1,000 Records.

The new Stream WIT leaves `append`, `close` and `read` out of the Workflow target and gives the Producer parameters their default. Its Rust code is the same, so the bridge needs no rebuild.

It brings the Workflow Stream without append, close and read from the generated-code branch.

The regenerated Workflow `Stream` lost the three methods, but Workflow code can still start their operations through a Nexus client, so the Worker keeps its refusal and the tests start them that way. The generator script names the nexgen commit that made the code.

The generated Workflow modules import `temporalio.workflow` at their top, and it imports them back, so a new interpreter that started with one of them failed. The parent package now loads `temporalio.workflow` last, which completes it before a generated Workflow module starts.

An Activity interceptor sees every Activity that runs on its Worker, also one that the SDK schedules. A tracing interceptor then shows the Store call under its Workflow, which is correct.

This release has no Streams on the server, so such a call could only fail late, with a message that didn't say why. It now fails at once with a `StreamConfigError` that names the fix.

A person who writes a Store for another system needs typed operations, not an operation name and bytes. Bytes now belong to the Stores on the bridge, and the hooks hand messages to a Store with no extra parse.

Worker code should only find a Store and hand it the call, not know how the Store serves Workflow code. Each Store now brings its own Activities and its own call from Workflow code, with a default Local Activity that carries the Store's name.

The crate writes its test vectors with a Rust program, so the submodule has no Python file. Its Rust code is the same, and the bridge needs no rebuild.

The submodule has no Python file any more, so pyright, pydocstyle and ruff have nothing to skip there.

A Store is an interface that a person can implement, so the docstring no longer calls it configuration.

The Workflow Task info of a Local Activity has a branch of its own below the generated code, so this change no longer shows it.

moetemp · 2026-11-14T01:51:30Z

closed in favour of temporalio#2129

moetemp changed the base branch from moe/brook/review/base to moe/streams/restart-base November 13, 2026 02:44

moetemp force-pushed the moe/brook/streams-combined branch from a53e75c to 7a643ba Compare November 13, 2026 02:44

moetemp added 6 commits November 12, 2026 20:38

Pointed Core at a commit that vendors the Stream API.

ccd0e97

sdk-python takes its API files from Core's vendored copy. That commit adds the Stream protos and WIT from the api repo until the api Stream change merges.

Generated the Stream protos, bindings, visitor and Workflow exports.

02645de

Written by `poe gen-protos` with NEX_GEN_BIN at nexgen 0.2.8 and NEX_GEN_STREAM_BIN at a build of nexgen commit bf3c49c9. Nothing in this commit is edited by hand.

Added the support functions that the generated Stream code calls.

7f2eca0

The generated code calls these by name for request ids, Producers and read paging. The system Nexus serialization lookup also knows the Stream operations.

Routed a client's Stream operations to a Stream Store.

20815db

Client inherits the generated extension, so its Stream calls are the generated ones. A Store answers them in place of the server, with the conversion and the codec of the system Nexus path.

Ran Workflow Stream operations in a Local Activity with a Store.

a92a582

Workflow code can't do I/O, so a Worker whose client has a Stream Store calls it from an internal Local Activity. The Replayer takes the Stores too, so a replay makes the same commands.

moetemp force-pushed the moe/brook/streams-combined branch from a5d7b87 to a92a582 Compare November 13, 2026 04:20

moetemp added 7 commits November 12, 2026 21:44

Gave an Activity's info the Workflow Task of a Local Activity.

0c0f2e1

Core sends the Workflow Task that dispatched a Local Activity. A write from a Local Activity needs it for its Producer epoch.

Wrote a Workflow's Stream append as its Local Activity's Producer.

e4c2b32

On a Store, a Workflow appends as a Local Activity Producer, and the Store refuses the Workflow Producer. Its epoch is the Workflow Task that ran the Local Activity, so the Store fences the write of a failed task.

Pointed Core at the commit with the Stream crate.

a644cf5

The bridge links the Stream runtime from Core's new `temporalio-streams` crate. The Core commits add no proto that the SDK generates from.

Bound the Stream runtime in the bridge behind a streams feature.

1f154d7

A Store on the bridge needs the crate's runtime and its memory Store. The feature is off by default, and the build tasks and wheels turn it on.

Raised one StreamError for every failed Stream call.

1f0993a

Client, Activity and Workflow code can catch one class and match on its reason. The details carry the reason, so Workflow code rebuilds the same error from a Local Activity's failure.

Added a memory Stream Store and moved the Stream tests to it.

43654eb

Tests now run on the same Stream runtime and rules as every Store on the bridge. They check Records and Producers with the generated read, so the fake Store is gone.

Gave a Workflow's Stream create the chain's first run id.

7337df1

Each run of a chain sends the same strict create. With one request id for the chain, the Store answers a later run's create as created and not as ALREADY_EXISTS.

tconley1428 reviewed Nov 13, 2026

View reviewed changes

moetemp added 6 commits November 13, 2026 12:52

Moved the Worker side of a Stream Store into one module.

1c836c8

The Activity, the Local Activity call and its handle were spread over generic Worker files. One module now holds all that a Worker does for the Stream Store of its client.

Made the call of a Stream Store private.

f1fdc70

A Store is the Rust runtime behind the bridge, so no user writes one in Python. Its call takes the bridge's operation name and bytes, which is not an API for users.

Said why stream_stores is a list, and named the Redis step in CI.

ed82555

A reader asked why a list takes one Store. A named CI step shows why Redis is installed.

Let WorkflowEnvironment take a Stream config for its client.

fa62ab2

A test that starts its own server needs a client with Stores, as a test with `Client.connect` has. `start_local` and `start_time_skipping` give the config to the client that they build.

Let a Worker serve every Stream Store of its client.

f1629bb

A Workflow's Stream call runs in the SDK's Local Activity, so the Store can be picked there by the client's rules. History keeps the ref with the Store's name, so a replay calls no selector.

moetemp requested a review from tconley1428 November 13, 2026 20:50

moetemp added 15 commits November 13, 2026 14:06

Moved the options of the Stream Local Activity to the Stream config.

dfa007b

A Workflow's create picks its Store only in the Local Activity, so with several Stores it could not use the options of any one Store. The config now holds one set of options for every Store.

Read the ref of a Stream call from its request object.

2d86f5b

Decoding the encoded request to find its ref decoded every Record of an append once more. The caller has the request object before it encodes it, and the Local Activity gets the ref as two strings.

Added BufferedAppender, which batches the appends of one Stream.

d0fba10

An Activity that streams model tokens appends a few characters many times a second, and one Store call for each is too many. The appender writes them in batches of at most 1,000 Records.

Moved Core to the API files with the Workflow Stream marks.

51a7d0b

The new Stream WIT leaves `append`, `close` and `read` out of the Workflow target and gives the Producer parameters their default. Its Rust code is the same, so the bridge needs no rebuild.

Merged the regenerated Stream code.

ea1523d

It brings the Workflow Stream without append, close and read from the generated-code branch.

Let the user's Activity interceptors see the Stream Local Activity.

56c9937

An Activity interceptor sees every Activity that runs on its Worker, also one that the SDK schedules. A tracing interceptor then shows the Store call under its Workflow, which is correct.

Raised an error for a Stream call when the client has no Store.

63180a3

This release has no Streams on the server, so such a call could only fail late, with a message that didn't say why. It now fails at once with a `StreamConfigError` that names the fix.

Gave StreamStore the five Stream operations as methods.

6427a7f

A person who writes a Store for another system needs typed operations, not an operation name and bytes. Bytes now belong to the Stores on the bridge, and the hooks hand messages to a Store with no extra parse.

Moved Core to the Stream crate that has no Python script.

419fc75

The crate writes its test vectors with a Rust program, so the submodule has no Python file. Its Rust code is the same, and the bridge needs no rebuild.

Dropped the lint excludes for the Core submodule.

a69aef6

The submodule has no Python file any more, so pyright, pydocstyle and ruff have nothing to skip there.

Removed a stale sentence from the StreamStore docstring.

138f338

A Store is an interface that a person can implement, so the docstring no longer calls it configuration.

Merged the generated-code branch, which sits on the Workflow Task info.

51fe79d

The Workflow Task info of a Local Activity has a branch of its own below the generated code, so this change no longer shows it.

moetemp closed this Nov 14, 2026

Conversation

moetemp commented Nov 9, 2026 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

moetemp commented Nov 14, 2026

Uh oh!

Reviewers

Uh oh!

Assignees

Labels

Projects

Milestone

Development

Uh oh!

2 participants

moetemp commented Nov 9, 2026 •

edited

Loading