[moetemp]

This PR adds three methods that scan a Stream, take, filter and reduce, to the client Stream and the Workflow Stream. It also adds temporalio.contrib.streams.pipeline, which chains steps over a Stream and runs the chain as one of these calls. It is a prototype, created for the purpose of getting your feedback on the API. It sits on #2129.

@workflow.defn
class Chat:
    @workflow.run
    async def run(self, ref: StreamRef[Token]) -> str:
        tokens = workflow.get_stream(ref, body_type=Token)

        # Wait for the next tool call, for at most a minute.
        calls = await tokens.filter(
            lambda record: record.body.kind == "tool_call",
            timeout=timedelta(minutes=1),
            max_records=1,
        )

        # Join the text of the tokens after it into the answer.
        answer = await tokens.reduce(
            lambda text, record: text + record.body.text,
            "",
            timeout=timedelta(minutes=1),
            from_offset=calls.last_offset,
        )
        return answer.value

Where to look

Part	Files	What it has
Generated methods	temporalio/nexus/system/stream_service/	take, filter and reduce on both Stream classes, and the models StreamRead, StreamReduce and StreamReadStop.
Helpers	temporalio/nexus/system/_support.py	take_stream, filter_stream and reduce_stream. Each generated method calls one. A helper checks the call and sends it to the path of Workflow code or to the path of a client.
The scan	temporalio/client/_stream_read.py	The loop that reads pages until a stop, the result that History keeps, and the Local Activity of a Workflow call.
Workflow side	temporalio/worker/_stream_read.py, _workflow_instance.py, _stream_store.py	Runs a call of Workflow code as one Local Activity, and checks its result on a replay.
The replay setting	temporalio/client/_stream_store.py, temporalio/exceptions.py	StreamConfig.workflow_read_replay_from, StreamReplaySource and StreamReplayError.
Pipeline	temporalio/contrib/streams/_pipeline.py, README.md	pipeline, with the steps filter, map, take_until and limit, and the ends records, reduce, collect and count.
Appender	temporalio/contrib/streams/_appender.py	BufferedAppender inside a function of a call.
Generation	scripts/gen_nexus_system_api.py, .github/workflows/ci.yml, temporalio/bridge/sdk-core	The nexgen pin and the Core pin that have the three methods.
Tests	tests/nexus/test_stream_read_workflow.py, test_stream_read_client.py, test_stream_read_replay_from.py, tests/contrib/streams/test_pipeline.py	Each stop, a replay, a payload codec, the three calls from a client and an Activity, the replay setting, and the pipeline.

How it behaves

• The three calls. take gives every Record up to a stop. filter gives the Records that its predicate keeps, and with max_records=1 it gives the first match. reduce folds the Records into one value. All three take the same optional parameters in the same order. Every result has last_offset, so a loop loses and repeats no Record.
• Stops. StreamReadStop says why a call ended. Its cases are Match, Done, MaxRecords, MaxBytes, Idle and Timeout, in that order when several hold. idle_timeout is 1 second by default, max_records is 1,000 and max_bytes is 256 KiB. timeout is required.
• Workflow code. A call runs its scan in one Local Activity, and History gets one marker with the result. For take and filter the result has the kept Records with their bodies, and for reduce it has the value. A replay takes the result from History, so it calls no Store. A result over 1 MiB fails the call.
• Records out of History, by a setting. StreamConfig(stores=[...], workflow_read_replay_from=StreamReplaySource.STORE) keeps the Records of every take and filter of Workflow code out of History. The marker then has the offsets of the kept Records, a count and a digest. Before a replay gives the call its result, the Worker reads those Records from the Store again and checks the count and the digest. So the Records must stay in the Store for as long as the Workflow can replay. A replay that can't read them fails the Workflow Task with StreamReplayError. A replay follows what the marker has, not the setting, so a change of the setting doesn't break a running Workflow. reduce keeps its value in History under both settings.
• Client and Activity code. A call runs the same scan in the caller's process. Each page is one read of the client Stream, so the client's interceptors see it. The result has no size limit, since it doesn't go into History.
• The functions of a call can't be serialized. In Workflow code they reach the Local Activity through a registry in the Worker process. So they run on the Worker's event loop and must change no state. A coroutine function is refused.
• max_bytes counts the encoded size of each kept Record, in every context. A typed page of read has no sizes, so the client's Store call hands them to the scan through a context variable.
• BufferedAppender in a function. A function can append to another Stream through a BufferedAppender. The call writes the appender's Records before it returns, so a fold can transform one Stream into another with one marker.
• The pipeline. Each step gives a new chain and runs nothing. An end runs the whole chain as one take, filter or reduce, so in Workflow code a chain is one Local Activity and one marker. The steps keep no state of their own, so a Local Activity that runs again gives the same result.

It needs no code change in Core or in the temporalio-streams crate.

What it costs, and its limits

• A take of 40 small Records is one Workflow Task and one marker of about 5.5 kB. A filter that keeps 5 of the 40 is about 1 kB. With STORE, the two markers are 383 and 518 bytes.
• A call keeps its Workflow Task open until it returns, so a Signal, an Update or a cancel reaches the Workflow only then. The default idle_timeout bounds that on a quiet Stream, and a short timeout, such as 2 seconds, bounds it while Records flow. A call that runs longer than 80% of the Workflow Task timeout makes them wait for a Workflow Task heartbeat.
• Workflow code doesn't run for each Record. A function of a call does, but in the Local Activity.
• What a call takes goes into History. So Workflow code should take only what it decides on. filter and reduce keep the bulk of a token Stream out.
• With STORE, a call has no default max_bytes and no 1 MiB limit, since no body goes into History. A replay of a filter makes one Store read for each kept Record.

What it depends on

• nexgen. The Stream WIT marks the three methods with @nexus.helper, which only the fork moetemp/nexgen has yet. So poe gen-protos and the check-protos job build the Stream package with a commit of that fork. The pin moves back to nexgen's main once the helper methods are there.
• Core. The submodule moves to a commit of moe/streams-crate-for-sdk-python whose only change is the Stream WIT with the three methods, as in feat(brook): add the Stream service messages (as experimental) api#895.

Open points

• The generated methods get their docstrings from the WIT. So the Python details (History, the Local Activity, the limits) are only in the docstrings of the helpers. Where should a user read them?
• The Local Activity of a Workflow call reads its Store directly, so a client interceptor doesn't see those pages.
• temporalio.contrib.streams sits next to temporalio.contrib.workflow_streams. Is the name clear enough?
• The Redis tests don't cover a replay under STORE yet.
• Is timeout the right name, or should it be max_duration?