[moetemp]

What was changed

This PR adds Workflow Streams (workflow.create_stream, create_stream_nowait, stream_handle and WorkflowStream), which write through Local Activities.

• Each create and publish is a Local Activity with the epoch of its Workflow Task (G3), so the Store withdraws the writes of a failed task when its retry writes.
• A create is create-or-get by the Stream id, <workflow id>:<first run id>:<name>, with an optional backend name. So a task retry, a retry run and the run after continue-as-new get the same Stream. A publish to a Stream that nobody created raises StreamNotFoundError.
• A close is a regular Activity, which runs only after its task commits. So a task that fails first leaves the Stream open for its retry.
• Publishes don'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.
• Records convert with the Workflow's payload converter with no serialization context, as in Activities and clients. A close failure converts with the Workflow's failure converter.
• A write in read-only code, such as a query handler or an update validator, raises temporalio.workflow.ReadOnlyContextError. In a validator it fails the Workflow Task, as other read-only actions do.
• History keeps a close's result and failure as the input of the close Activity, so the payload size limit applies to them. They stay there, since only a regular Activity runs after its task commits. To move them out, a Store needs a call that keeps a close result until the close, and no Store has one yet.
• Continue-as-new passes the run chain's open Streams on. A Workflow with no Stream never imports the Streams package.

Depends on S6 (#376), and on G3 (#369) through feature-streams. No changelog fragment, since S9 has the Streams entry.
Part of the Brook review series.

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 handle, the Local Activities and the interceptors only test together. If you want it smaller, I can split it in two.

• S7a. Create, publish and close through Local Activities, with the Activity interceptor and the write tests.
• S7b. The block-end close when the run has no retry run after it, and continue-as-new (close_at_run_end, _take_owned_streams and their tests). About 300 lines.

Open question. Workflow Streams use private parts of the Workflow runtime. This is the hook API that would replace them. Should the SDK add it?

• workflow.unsafe.run_local(key, factory) gives state that lives as long as the run. It replaces the attribute on _Runtime.current().
• workflow.unsafe.outbound_interceptor(cls) gives the run's outbound interceptor of that class, or None. It replaces the walk of the _outbound chain.
• workflow.unsafe.fail_run(error) fails the run with error after the current callbacks, as if the run raised it. It replaces _current_activation_error, which the error of a _nowait call sets.
• workflow.unsafe.is_failure_exception(error) tells if error fails the run, and not only the Workflow Task. It replaces _Runtime.workflow_is_failure_exception.
• workflow.unsafe.payload_converter(context=False) and workflow.unsafe.failure_converter(context=False) give the converters with no serialization context. They replace _context_free_payload_converter and _context_free_failure_converter.

A publish future sets a flag when a caller gets its result or its error, so the code doesn't read Future._log_traceback.

Checklist

1. How was this tested:
   • poe lint, poe gen-docs and pytest tests/contrib/streams on both backends, with Redis 8 and Redis 7.0 (CI's version on Ubuntu). Fork CI passes, except the arm jobs, since the fork has no arm runners.