Add a public workflow-side read API to workflow streams

[DABH]

What was changed

temporalio.contrib.workflow_streams.WorkflowStream (experimental) gains a public workflow-side read API:

• WorkflowStream.base_offset — global offset of the oldest entry still retained in the log. Starts at 0, advances when truncate() discards a prefix, and carries across continue-as-new via prior_state.
• WorkflowStream.end_offset — global offset one past the newest entry (base_offset + retained length); the same value the offset query reports. Monotonic, so it composes with workflow.wait_condition predicates.
• WorkflowStream.read(topics=None, from_offset=0, *, result_type=None) — a deterministic, non-blocking snapshot of the retained entries at global offsets >= from_offset, each returned WorkflowStreamItem carrying its global offset. Semantics deliberately mirror the existing public surfaces:
  • topic filtering and result_type decoding identical to WorkflowStreamClient.subscribe, including the result_type=RawValue escape hatch and the result_type=Payload rejection;
  • truncation semantics identical to the poll handler: from_offset=0 means "from the beginning of whatever is retained", while a non-zero offset below base_offset raises an ApplicationError of type TruncatedOffset.

The change is purely additive — no behavior changes to the publish/poll/query paths. Docstrings for WorkflowStreamItem and WorkflowTopicHandle were updated to reflect that a workflow-side read path now exists, and a changelog entry was added.

Consuming entries as they land, from workflow code:

cursor = self.stream.end_offset
while not done:
    await workflow.wait_condition(lambda: self.stream.end_offset > cursor)
    for item in self.stream.read(from_offset=cursor):
        ...  # item.offset is the item's global offset
    cursor = self.stream.end_offset

Why

The write and subscribe paths of workflow streams are public in both directions (workflow-side topic().publish(...), client-side publish and subscribe), but there is no public way for workflow code to read the stream's entries back. Integrations that dispatch streaming work to an activity and consume the streamed items inside the workflow — for example, agent-framework middleware that re-yields streamed activity output into the framework's own async-generator machinery — currently have to reach into the private _log / _base_offset attributes to do that.

A concrete consumer exists: a NeMo Agent Toolkit durability plugin that publishes each streamed item from an activity as an enveloped record on a topic and consumes exactly its own invocation's items, in sequence order, from workflow code, with truncation detection. Its requirements shaped this surface: read entries from an offset, learn the truncation-aware base offset, and iterate as new entries land in a workflow.wait_condition-compatible way. That consumer's wake predicate also incorporates activity-handle completion, which is why the offset primitives are exposed (enabling arbitrary wait conditions) rather than only a packaged iterator; a convenience iterator can be layered on later if there is demand.

Testing

• New e2e test (test_workflow_reads_activity_published_items): a workflow consumes items an activity publishes to its own stream via the documented end_offset + wait_condition + read loop, running with max_cached_workflows=0 so every activation replays the read loop from history (exercises determinism under replay).
• New in-workflow probe test (test_workflow_read_offsets_filters_decode_and_truncation): global offsets with and without topic filters, suffix and empty reads, base_offset/end_offset before and after truncate, default and RawValue decoding, the result_type=Payload rejection, and the TruncatedOffset error path.
• Full tests/contrib/workflow_streams/ suite passes (46 tests), including against the time-skipping test server.
• pyright, mypy, basedpyright, pydocstyle, ruff format/import checks, and gen-docs are all clean.

[DABH]

Withdrawing this: the consuming integration now implements workflow-side reads over the public get_state() snapshot (decoding the wire entries directly), so no new API surface is needed for this use case. If more in-workflow consumers appear, the design notes here are a starting point — with the naming caveat from review that a waiting iterator should own the subscribe name and read is reserved for post-close retrieval.