[moetemp]

This PR adds Workflow Streams (workflow.create_stream, create_or_get_stream, their _nowait forms, to_stream and WorkflowStream, which makes the Stream calls in Workflow code). They write through Local Activities.

• A call takes the payload type as a type argument, as in workflow.create_stream[Turn]("session-7-turns"). A Workflow creates only standalone Streams, under an id of its choice, so the ref has no owner and the Stream doesn't end with the run.
• create_stream fails with StreamAlreadyExistsError if the id exists, and create_or_get_stream gets it. The request id is <first execution run id>:<stream id>, so it's the same in each run of the chain. A Workflow Task retry, a retry run or a cron run that creates the Stream again gets the Stream that the first create made. A client sends a new request id with each create call. In a Workflow the request_id() source of [Brook S2] Generated the Stream service bindings with nexgen, and added the writer client. #372 now returns an empty id, so a create draws nothing from the user's workflow.random().
• A run after continue-as-new uses the earlier run's Streams through the refs that it forwards, with workflow.to_stream. There's no workflow.get_stream.
• Workflow code calls the generated Workflow API of [Brook S2] Generated the Stream service bindings with nexgen, and added the writer client. #372. A Workflow outbound interceptor gets those calls through start_system_nexus_operation and runs each one as a write Local Activity. So History keeps a marker per write and has no Nexus events.
• Each publish is a Local Activity whose Producer epoch is the local_activity case with the Workflow Task that dispatched it ([Brook G3] Added activity.Info.workflow_task for Local Activities. #369). So a reader withdraws the writes of a failed task once its retry writes. A retry run's first publish to a Stream withdraws the run that it retries, and a create never does, since the Store refuses that on a Stream with no owner. The workflow_writer() and client_writer() sources of [Brook S2] Generated the Stream service bindings with nexgen, and added the writer client. #372 give the run's writer and the Activity attempt's writer.
• A close carries no value. It's a regular Activity, which runs only after its task commits and closes as the run's Workflow Producer. So a task that fails first leaves the Stream open for its retry.
• A write takes its place in the Stream's order at the call, since _support.start_now runs the generated call up to its first wait. So publish_nowait, close_nowait and the _nowait creates work in sync code. The error of a _nowait write fails the run once no other write is in flight, since no Local Activity marker may follow the fail command. A write queued behind a create that fails sends nothing and fails with the create's error, since the Stream can be another owner's.
• publish returns a future and doesn't block the Workflow. A publish error that no caller retrieved fails the run at its end, and the Workflow's logger logs it when the write fails.
• The end of the run waits for every write, also for a write that a Signal or Update handler starts while it waits. A cancel there doesn't stop the writes. The run waits for them, then ends as canceled.
• async with closes at the block end, also when the run continues as new in the block. It defers the close to the run's end when a retry run may follow. A BaseException, such as an interrupt, follows the same rule as an exception. On a Stream bound to another run chain it raises StreamNotOwnerError at __aenter__.
• Records convert with the Workflow's payload converter with no serialization context, as in Activities and clients.
• A write in read-only code, such as a query handler or an update validator, raises temporalio.workflow.ReadOnlyContextError.

Why?

This is the main use of Streams. An agent Workflow writes its turns to a Stream, and a client or a UI reads them as the Workflow writes them.

This PR is over the size target, since the Workflow calls, the Local Activities and the interceptors only test together.

Open question. Workflow Streams use private parts of the Workflow runtime, such as the run's outbound interceptor chain, the context-free converters and the error that a _nowait call raises in the run. Should the SDK add a hook API in workflow.unsafe for them?

Part of the Streams review series.

Checklist

1. How was this tested:
   • poe lint, poe gen-docs and pytest tests/contrib/streams on both backends.