Added the workflow publish and subscribe commands.

[moedash]

This PR gives a workflow the commands to publish to and subscribe to a stream.

AppendStreamRecords appends to a stream the workflow owns, which is a co-located subcomponent, so the batch and the frontier land in the workflow task's own commit. SubscribeStream records a cursor. Both write an offsets-only event.

What changed?

• chasm/lib/workflow/stream_commands.go: the two command handlers, their event definitions and the library that registers them. A publish clears the producer id, which is how a reader tells the owning workflow's records from an outside producer's.
• The events carry offsets and nothing else. A batch of one 20-byte record and a batch of a thousand 2KB records write the same event, because the bodies stay in the stream component. They exist because every SDK matches issued commands against command-generated events in order, and a command that produces none puts that matching out of step.
• A refused publish or subscribe fails the workflow task with its own cause instead of failing the completion call, which would leave the worker retrying the same command with nothing in History saying why.
• A subscribe is staged rather than resolved in the handler: a stream in another execution has to be pinned on its own shard, which a command handler holding the state lock cannot reach. respondworkflowtaskcompleted/stream_appends.go resolves the staged subscriptions between the commands and the commit, through the routed stream client. The pin goes on the stream before the cursor goes on the workflow, so an interruption leaves a pin nothing reads rather than a cursor no truncation floor protects.
• A subscribe that names neither an owned stream nor a standalone one creates the owned stream. A workflow reads a topic before anything has been written to it, and the name it subscribes with is the stream an outside producer later appends to.
• service/history/history_engine.go installs the routed stream client on the workflow task paths; service/history/configs/config.go carries the namespace limits into them.
• tests/stream_workflow_test.go and tests/stream_rejection_test.go prove publishing end to end, that a refused publish fails the task with a cause and appends nothing, and that a publish on a failed task leaves no event, no offset and no message behind.

Why?

The publish path is the one that makes the design worth having: it costs one event per batch and it is atomic with the workflow task. That is easier to judge on its own than mixed with the delivery machinery.

How did you test it?

go build ./...
make fmt              # tree stays clean
go test ./chasm/lib/... ./service/history/... -count=1
go test -tags test_dep ./tests/ -count=1 -run \
  'TestStreamWorkflowPublishesWithARangeEvent|TestStreamWorkflowLongPollWakesOnPublish|TestStreamWorkflowTakesAppendsFromOutsideToo|TestStreamWorkflowStreamClosesWithItsWorkflow|TestOutsideAppendsRaceTheWorkflowPublishWithoutFailing|TestOwnedStreamRefusesAppendsPastItsBudget|TestOverLimitPublishFailsTheWorkflowTask|TestPublishOnAFailedTaskLeavesNoTrace'

All green.

tests/stream_replay_test.go, tests/stream_consume_test.go and tests/stream_consumer_lifecycle_test.go arrive here holding only the helpers the tests above share. They gain their own tests in the next layer, which is where a subscription starts being delivered to.

Series

Server 4 of 6. Replaces #2.

Previous: #5 moe/AI-198-srv-3-stream-service. Next: #7 moe/AI-198-srv-5-delivery-replay.

Review round

A subscribe no longer reads any NotFound from the routed RegisterStreamConsumer as proof that the id names one of the workflow's own streams. The stream's own shard answers stream_absent for that one case, and a registry miss or a shard that has moved stays an error, so a routing failure can no longer bind the workflow to a different, empty stream with a History event that looks identical to the intended subscription. A refusal from SubscribeToExternalStream now fails the workflow task with a cause like every other refusal in the same loop, instead of failing the completion call.

Both commands answer stream.enabled, since the command path does not go through the stream service and the frontend gate does not cover it. Subscriptions per execution are bounded by stream.maxSubscriptionsPerWorkflow, counting what the current task has staged, because each one costs a routed call on the completion path and another on every task start, both with the execution's lock held. The whole set of those calls shares one deadline, stream.RoutedSetBudget, so the lock hold no longer grows with the subscription count. A publish counts against the shared byte budget of every stream the execution owns.

Known limits, argued rather than fixed:

• A subscription to a stream the workflow owns ends at a continue-as-new. The stream lives in the execution and the execution does not survive the run transition, so the successor gets a new stream of the same name starting at offset zero, and records written to the predecessor's between the commit and a re-subscribe are not carried. TestOwnedStreamSubscriptionEndsAtContinueAsNew in Delivered stream records on Workflow Tasks and on replay. #7 pins the behaviour. A subscription to a stream in another execution does carry.
• StreamCursors still keys an owned stream by name and a standalone one by id in one map, and StreamRange.stream_id inherits that. The collision is refused rather than resolved, so nothing binds to the wrong data, but History does not say which origin a range came from. Qualifying the key is an api change and belongs with the field rename.

Review round: api surface

The api branch renamed AppendStreamRecordsCommandAttributes.stream_id to stream_name and SubscribeStreamCommandAttributes.stream_id to stream_name_or_id, and gave WorkflowStreamRecordsAppendedEventAttributes a from_offset and a to_offset in place of a first offset and a count. Field numbers are unchanged, so the wire does not move. This branch owns the command handlers, so it is where the reads change. handleAppendStreamRecordsCommand resolves GetStreamName(), which is what it always did with the value. handleSubscribeStreamCommand reads GetStreamNameOrId(), which is honest about a value the server resolves as an owned name first, then as a standalone id, and failing both creates an owned stream from. RecordStreamRecordsAppended now takes the two offsets and passes the batch's NextOffset straight through rather than a count.