[moetemp]

What changed?

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's a prototype, created for the purpose of getting your feedback on the API. Its base is #2129.

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

• The three calls. take gives every Record up to a stop. filter gives the Records that its predicate keeps. reduce folds the Records into one value. Each result has last_offset for the next call, and a stop that says why the call ended.
• Workflow code. A call runs its scan in one Local Activity, and History gets one marker with the result. A replay takes the result from History and calls no Store. A result over 1 MiB fails the call.
• Records out of History, by a setting. With workflow_read_replay_from=StreamReplaySource.STORE on the StreamConfig, the marker of a take or a filter has only offsets, a count and a digest. A replay then reads the Records from the Store again, and it fails the Workflow Task with StreamReplayError if it can't. So the Records must stay in the Store for as long as the Workflow can replay.
• Client and Activity code. The same scan runs in the caller's process, and the result has no size limit.
• The functions of a call can't be serialized. In Workflow code they run in the Local Activity, on the Worker's event loop, and must change no state.
• The pipeline. A step, such as filter or map, runs nothing. An end, such as records or count, runs the whole chain as one call. So in Workflow code a chain is one Local Activity and one marker.

It depends on two things that have no release yet.

• 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.
• Core. The submodule points to a commit whose only change is the Stream WIT with the three methods, as in feat(brook): add the Stream service messages (as experimental) api#895. Core's code doesn't change.

Why?

The Workflow Stream of #2129 has no read. Workflow code usually doesn't need each Record. It needs a decision from the Stream, such as the next tool call or the full text of an answer. These calls give that decision with one Local Activity and one marker, and Workflow code doesn't run for each Record.

It has two costs.

• 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 of 1 second bounds that on a quiet Stream, and a short timeout bounds it while Records flow.
• What a call takes goes into History. A take of 40 small Records is one marker of about 5.5 kB, or 383 bytes with STORE.

These points are open.

• The generated methods get their docstrings from the WIT, so the Python details (History, the Local Activity, the limits) are only on 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.
• Is the package temporalio.contrib.streams clear enough next to temporalio.contrib.workflow_streams?
• Is timeout the right word, or should it be max_duration?

How did you test it?

The tests are in tests/nexus/test_stream_read_*.py and tests/contrib/streams/test_pipeline.py. They run on the local dev server with the memory Store. The Redis tests don't call the three methods yet.

Link to a test plan if any -

• Unit Tests
• Staging
• End to End Tests