Added server-side streams for workflows and external clients.

[moedash]

What changed?

This PR adds server-side streams, the Option 5 prototype. A stream is a CHASM component whose payload lives in the component's data nodes. The stream service RPCs land with frontend wiring. A workflow appends records with a command that commits inside its Workflow Task, and a consuming workflow gets consumed ranges delivered on the task, routed across History hosts. Retention refuses to trim below an active consumer, and a capped stream refuses the append instead of deleting the range. The append dedup fingerprint is independent of protobuf map order. The benchmark suite sits behind TEMPORAL_STREAM_BENCH=1.

A subscribe command naming a stream on another History host is registered through the routed stream service, so it works on a cluster with more than one History host. A refused append or subscribe fails the Workflow Task with WORKFLOW_TASK_FAILED_CAUSE_BAD_APPEND_STREAM_RECORDS_ATTRIBUTES or WORKFLOW_TASK_FAILED_CAUSE_BAD_SUBSCRIBE_STREAM_ATTRIBUTES instead of failing the completion call. A task whose recorded stream range can no longer be served, or whose cold replay would exceed the re-supply budget, fails with WORKFLOW_TASK_FAILED_CAUSE_STREAM_RANGE_UNAVAILABLE and a message naming the stream. Cold replay follows every page of history to find the recorded ranges rather than the first page only. A record appended with no kind is stored as data. An RPC append no longer pins to the head it read; only an explicit expected_offset is checked. A stream a workflow owns carries a byte and item budget, and the resource-bounding limits are namespace-scoped dynamic config under stream.*. A consumer's pin on a stream is keyed by run, released when the run closes, and moved to the successor after continue-as-new. Retention waits while a consumer is active, and DeleteStream refuses while one is unless force is set. The frontend checks stream ids, names, offsets and page sizes before routing. Every slice names the run that holds the stream.

The wire follows the record vocabulary of the api: AppendStreamRecords, WorkflowStreamRecordsAppended, consumed_stream_ranges and StreamRange. Each StreamRecord carries its kind, producer_id, attempt and sequence; the server stores them as sent and delivers them on the Workflow Task, and the workflow's own appends are stored with an empty producer_id. A subscribe naming a stream this workflow does not own yet, and that exists nowhere else, creates the owned stream so a reader can subscribe before the first record arrives.

The stream service declares its authorization metadata. Every method is namespace-scoped. PollMessages, PollWorkflowMessages, DescribeStream, DescribeWorkflowStream and ListStreams are read-only. CreateStream, AddMessages, FinishWriting, SubscribeWorkflow, AddWorkflowMessages, CloseStream, TruncateStream and DeleteStream are writes. RegisterStreamConsumer and AdvanceConsumerHead, the two calls History makes on itself, are admin, and the frontend answers them with Unimplemented anyway. Whether closing, truncating and deleting a stream should need an operator rather than a namespace writer is a policy call this PR leaves open. Every method has a rate-limit priority next to the workflow service's: appends and subscribes with the signal-like APIs, close, truncate and delete with the state changes, the describes with the status reads, the two polls with the other polls and counted as long-running, ListStreams with the visibility reads. The redirection interceptor used to serve a stream call wherever it landed; it now forwards a call for a namespace active in another cell to that cell, the way it forwards a workflow service call, through the interceptor's registered-method hook.

A query dispatched straight through matching now carries the recorded stream ranges. Matching asks History for them through a new internal RPC, GetStreamReplaySlices, after it has fetched the history for a non-sticky query task, and History re-supplies them from the streams with the same page walk a cold Workflow Task uses. Every slice on a query task is tagged with the completion that recorded it. There is no untagged slice, because a query starts no task. A range the stream can no longer serve fails the query with FailedPrecondition, since there is no task to fail.

A workflow reset carries its subscriptions. A run rebuilt from its events recreates its cursors: the WorkflowStreamSubscribed event places a cursor, each completed task's consumed_stream_ranges move it, and a range for a subscription made out of band, which leaves no event, creates the cursor from the range. The reset then copies what only the base run knows: whether a stream lives in another execution, and the frontier that stream last pushed. A cursor on a stream the base run owned gets a stream of the reset run's own that begins at the offset the cursor stands at, so the ranges recorded before the reset point stay in the base run's stream and everything from the reset point on is the reset run's. Replay reads each recorded range from the run whose era recorded it, found by the WorkflowTaskFailed event with cause RESET_WORKFLOW and its base_run_id, and names that run in StreamSlice.run_id. The base run's pin on a standalone stream is released the way a closed run's is: the next notify finds the base run terminated and re-keys the pin to the reset run, at the floor the inherited subscription began at. The base run names its successor in DescribeWorkflowExecution under workflow_extended_info.reset_run_id.

Failover is characterized under both replication modes in tests/xdc/. With state-based replication (history.enableTransitionHistory) the CHASM nodes Streams#<name>, Streams$<name>$Batches#<offset> and StreamCursors#<name> reach the standby, the records and the frontier are readable there after the failover, and the workflow keeps consuming and publishing with the consumed range re-supplied on its first task. With event-based replication only what the events carry reaches the standby: the cursor is rebuilt from the recorded ranges, the stream and its batches are not, so the frontier reads as an unwritten stream and no record can be polled. Carrying the stream under that mode would take a replication task for the component, the way the HSM state replicator ships state; the appended event cannot carry it without putting payloads in History.

Fixes AI-198

Why?

Today's Workflow Streams spends History and Update budgets per record. With the payload kept out of History (an append is one fixed ~41-byte event), outside readers fan out without workflow tasks, and replay re-reads the recorded ranges.

Known limits

A history-only replayer, one that reads events without asking the server, has nothing to re-supply the recorded ranges from; the query path is covered now, the replayer is not. A consumer whose recorded range is gone for good keeps failing its task attempts until the stream is repaired or the workflow is terminated. After a reset, the base run's stream holds the records the reset run's history refers to, and the base run's retention eventually deletes them; a cold replay of the reset run after that fails its task with STREAM_RANGE_UNAVAILABLE. A reset run's own stream begins at the inherited offset, so a reader following the chain into it has to start at the floor DescribeWorkflowStream reports; polling below it is refused. A subscription made through the service rather than by a command leaves no event, so a reset run carries every such subscription the base run held, positioned where the reset run's history left it, even one made after the reset point. Under event-based replication the workflow's first task on the new cluster cannot find the stream its cursor names; the xdc test asserts what the standby holds and reads, not the stuck task. Taking the consumer pin before the workflow lock is held, rather than inside the completion transaction, is left for a later change. The Nexus payload commit duplicates upstream's fix and is dropped on rebase; the grpc and OpenTelemetry bumps come from the api-go module; the replace of go.temporal.io/api goes away once the api change is tagged.

How did you test it?

Link to a test plan if any -

• Unit Tests
• Staging
• End to End Tests

Unit suites plus the functional tests (tests/stream_*.go) on SQLite. Cross-host routing over the RPC and over the command, cold replay across history pages, refused commands, append atomicity on a failed task, consumer lifecycle across completion and continue-as-new, retention and deletion guards, subscribe-before-write and record identity on delivery have their own functional tests.

The authorization table is checked against every method of the generated service, the quota tables the same way, and the redirection of a stream call in the interceptor's unit suite; the redirection is not driven across two clusters. The query path has a functional test in which a poller that took part in no earlier task answers a query from the slices on the query task alone, and a matching unit test for the sticky and non-sticky cases; it is not exercised with an SDK worker. The reset has a functional test for an owned stream, including a cold replay of the reset run and the reads from outside, and one for the pin on a standalone stream; the rebuild and inheritance have unit tests, and a reset of a reset run is covered by a unit test of the page walk only. Two xdc suites fail a namespace over with a stream under state-based and under event-based replication.