[moetemp]

What changed?

This PR adds the Workflow module temporalio.contrib.streams.workflow, with create_stream, stream_handle, WorkflowStream and WorkflowTopic.

await create_stream(Events, *, name=None, retention=None, auto_close=None, close_with_run=True, backend=None) creates a Stream with the id <workflow id>:<first run id>:<name>. stream_handle(Events, *, name=None) gives the same Stream again with no call. await topic.publish(*values) puts the Records in the run's buffer and returns None. await stream.close(result=, failure=) goes in the buffer too, and async with closes on exit with the exception as the failure, with the existing cancel and BaseException rules.

WorkflowTopic.publish_nowait(*values) and WorkflowStream.close_nowait(*, result=None, failure=None) do the same with no await, for sync Workflow code such as @workflow.init and sync signal and update handlers, as asyncio.Queue.put_nowait does for put (S80). The awaited calls use them. create_stream_nowait takes the options of create_stream with no await, for a create in @workflow.init (S83), and create_stream calls it.

Workflow code only gets errors that replay reproduces: a Stream that this run closed, a Topic outside the definition, and read-only code. A Worker with no Stream plugin, or without the backend that the Stream names, raises a RuntimeError, so the Workflow Task fails and a Worker with the right config takes the run on. close_with_run=False on a Stream that already closes with the chain raises ValueError. close_with_run closes the Stream with the result or failure of the run chain's last run, through the existing close-with-run machinery. Continue-as-new passes the ids of the open Streams to close with the run chain to the next run in a header, so the last run closes them even when it does not use them. Each plugin takes name= for its backend (redis and memory by default) and default= to mark the client's default backend. It gives the name to the crate's config and the Workflow interceptor. The end of a retried chain now counts from workflow.Info.execution_expiration_time, a new field from InitializeWorkflow, since the execution timeout covers the whole chain.

Why?

S65, S68, S69, S72, S73, S76 and S77: Workflow code creates, writes and closes named Streams with awaited calls, and there is no default Stream.

How did you test it?

Live tests in tests/contrib/streams/test_workflow_module.py on the memory Store and on a scratch redis-server, against the stock dev server: create, publish of several values and close with the run's result, async with with and without a failure, continue-as-new with the handle in the input, the errors in Workflow code and in a query, Worker config errors that fail the Workflow Task, a failed run, auto_close, a cron chain, sync init, signal and update handlers that publish and close with no await, and a create and publish in @workflow.init. test_chain covers the expiry of a later run of a chain. The fixtures moved to conftest.py. pyright, mypy, basedpyright, pydocstyle, ruff.

• Unit Tests
• Staging
• End to End Tests

[moetemp]

Closing as stale. This layer is folded into the v5 tip, #365, and the upstream review uses a fresh series instead. Plan: https://www.notion.so/4048fc567738814990b7c796ac28fe3f