Add temporal-workflowstreams contrib module

[brianstrauch]

What was changed

Adds a new temporal-workflowstreams contrib module: a durable, multi-topic pub/sub log hosted inside a workflow, mirroring the workflow streams contrib packages in the Go, Python, and TypeScript SDKs. All APIs are marked @Experimental.

Wire protocol (cross-SDK contract): external publishers send batches via the __temporal_workflow_stream_publish signal, subscribers long-poll via the __temporal_workflow_stream_poll update, and the __temporal_workflow_stream_offset query exposes the current offset. The JSON envelope field names and the per-item payload encoding (base64 of the serialized temporal.api.common.v1.Payload) match the other SDKs exactly, so a Java publisher or subscriber interoperates with a workflow written in any of them and vice versa.

Workflow side: WorkflowStream registers a typed listener (preferably from a @WorkflowInit constructor) and supports publisher dedup (publisher ID + monotonic sequence), ~1 MB poll response paging, truncation, and continue-as-new state carryover via WorkflowStreamState.

Client side: WorkflowStreamClient (also constructible from inside an activity) provides a batching background publisher with retry and sequence-based exactly-once delivery, plus a blocking subscription iterator that follows continue-as-new chains, recovers from truncation, and ends cleanly on terminal workflow states.

Core SDK change: like the Go SDK (isWorkflowStreamReservedName), the signal/update/query dispatchers and listener registration now permit handler names in the exact __temporal_workflow_stream_ sub-namespace, which is otherwise reserved under the __temporal_ prefix. Other __temporal_ names remain blocked.

Why?

Brings the Java SDK to parity with the other SDKs' experimental workflow streams support, enabling durable event streams whose cost scales with durable batches rather than message count.

Checklist

1. Closes: n/a

2. How was this tested:

   • PayloadWireTest — wire-format round-trips (base64-of-proto, the cross-SDK contract)
   • StreamPublisherTest — publish-path unit tests with an injected signal function (batching, sequence advancement, force flush, drain-on-close, flush timeout)
   • WorkflowStreamTest — workflow-side integration tests against the in-process test server (external publish, dedup, topic-filtered polls, truncation errors, @WorkflowInit construction, custom payload converters)
   • SubscribeTest — subscription integration tests (delivery and offset advancement, topic filtering, truncation reset, clean terminal end, continue-as-new follow across runs)
   • WorkflowStreamReservedNameTest — reserved-name exemption works end-to-end; other __temporal_ names still rejected

3. Any docs updates needed: module README included (contrib/temporal-workflowstreams/README.md)

🤖 Generated with Claude Code

[diwu-sf]

@brianstrauch we would like to use this, our set up is:

• Temporary worker activity is using Python SDK, running the OpenAI agent sdk, events are published from this activity
• Java spring boot server provide SSE stream to web client
• Java need to be able to subscribe to the agent's LLM completion event stream to send to web client

Can you rebase this PR and get it fully merged?
Thanks

[diwu-sf]

One concern with the subscriber API: WorkflowStreamSubscription is currently a blocking Iterator, which means the application has to dedicate a driver thread to each active subscription.

The iterator loop is ergonomic:

try (WorkflowStreamSubscription sub = client.subscribe(options)) {
  for (WorkflowStreamItem item : sub) {
    ...
  }
}

but that caller thread is also what drives the long-poll loop. It blocks while waiting for poll update results, cooldowns, and terminal/continue-as-new handling. So the thread model becomes roughly:

one active workflow-stream subscription ~= one occupied driver thread

Could we expose a non-blocking subscriber API as the primary or additional API? For example:

WorkflowStreamSubscription<T> subscribe(
    WorkflowStreamSubscribeOptions<T> options,
    WorkflowStreamListener<T> listener);

interface WorkflowStreamListener<T> {
  CompletionStage<Void> onNext(WorkflowStreamItem<T> item);
  default void onError(Throwable failure) {}
  default void onCompleted() {}
}

With that shape, the implementation can run the blocking startUpdate(..., ACCEPTED) call on a shared executor, then wait for the long-poll result with getResultAsync() instead of keeping a caller thread parked for the duration of the poll. Many subscriptions can share the same executor/scheduler, and CompletionStage from onNext gives applications a backpressure boundary without blocking the poller thread.

The blocking iterator could still be useful as a convenience for synchronous consumers, but we would prefer not to make it the only subscriber interface, because it bakes in one driver thread per active subscription.

[Copilot]

Pull request overview

This PR adds a new temporal-workflowstreams contrib module that implements an experimental, workflow-hosted durable pub/sub log with a cross-SDK wire protocol, and updates the core SDK’s reserved-name handling to allow the __temporal_workflow_stream_ sub-namespace for the module’s signal/update/query handlers.

Changes:

• Introduces the contrib/temporal-workflowstreams module with workflow-side hosting (WorkflowStream) and client-side publishing/subscribing (WorkflowStreamClient) APIs plus protocol DTOs and internal drivers.
• Relaxes reserved-name validation/dispatch in the SDK to permit __temporal_workflow_stream_* handlers while keeping other __temporal_* names reserved.
• Wires the new module into Gradle settings, the BOM, adds README/docs, tests, and CODEOWNERS entries.

Reviewed changes

Copilot reviewed 39 out of 39 changed files in this pull request and generated 5 comments.

Show a summary per file

File	Description
temporal-sdk/src/test/java/io/temporal/workflow/WorkflowStreamReservedNameTest.java	Adds an end-to-end test for the reserved-name exemption for workflow streams.
temporal-sdk/src/main/java/io/temporal/internal/sync/UpdateDispatcher.java	Allows workflow-stream update names under the __temporal_ prefix to be dispatched/validated.
temporal-sdk/src/main/java/io/temporal/internal/sync/SignalDispatcher.java	Allows workflow-stream signal names under the __temporal_ prefix to be dispatched/buffered.
temporal-sdk/src/main/java/io/temporal/internal/sync/QueryDispatcher.java	Allows workflow-stream query names under the __temporal_ prefix to be dispatched.
temporal-sdk/src/main/java/io/temporal/internal/common/InternalUtils.java	Adds workflow-stream reserved prefix helper and updates reserved-name checking logic.
temporal-bom/build.gradle	Adds the new module to the published BOM.
settings.gradle	Includes the new contrib module in the multi-project build.
contrib/temporal-workflowstreams/src/test/java/io/temporal/workflowstreams/WorkflowStreamTest.java	Integration tests for workflow-side stream hosting (publish/poll/truncate/init behavior).
contrib/temporal-workflowstreams/src/test/java/io/temporal/workflowstreams/SubscribeTestWorkflows.java	Shared host workflow fixtures for subscription tests.
contrib/temporal-workflowstreams/src/test/java/io/temporal/workflowstreams/SubscribeTest.java	Integration tests for iterator-based subscriptions and rollover/truncation handling.
contrib/temporal-workflowstreams/src/test/java/io/temporal/workflowstreams/StreamPublisherTest.java	Unit tests for batching publisher behavior (flush, sequencing, retry/timeout).
contrib/temporal-workflowstreams/src/test/java/io/temporal/workflowstreams/PayloadWireTest.java	Tests for base64-of-proto payload wire encoding/decoding contract.
contrib/temporal-workflowstreams/src/test/java/io/temporal/workflowstreams/ListenerSubscribeTest.java	Integration tests for listener-based subscriptions, backpressure, and concurrency behavior.
contrib/temporal-workflowstreams/src/main/java/io/temporal/workflowstreams/WorkflowTopicHandle.java	Workflow-side per-topic publishing handle API.
contrib/temporal-workflowstreams/src/main/java/io/temporal/workflowstreams/WorkflowStreamSubscriptionHandle.java	Listener subscription handle API (close + done future).
contrib/temporal-workflowstreams/src/main/java/io/temporal/workflowstreams/WorkflowStreamSubscription.java	Iterator-based blocking subscription implementation backed by the shared driver.
contrib/temporal-workflowstreams/src/main/java/io/temporal/workflowstreams/WorkflowStreamState.java	Continue-as-new carryover state type with fixed JSON field names.
contrib/temporal-workflowstreams/src/main/java/io/temporal/workflowstreams/WorkflowStreamOptions.java	Workflow-side configuration options for payload conversion.
contrib/temporal-workflowstreams/src/main/java/io/temporal/workflowstreams/WorkflowStreamListener.java	Listener API for non-blocking subscription consumption and backpressure.
contrib/temporal-workflowstreams/src/main/java/io/temporal/workflowstreams/WorkflowStreamItem.java	Decoded subscription item type (topic/payload/offset).
contrib/temporal-workflowstreams/src/main/java/io/temporal/workflowstreams/WorkflowStreamHandlers.java	Reflective handler interface registering the protocol signal/update/query methods.
contrib/temporal-workflowstreams/src/main/java/io/temporal/workflowstreams/WorkflowStreamConstants.java	Cross-language protocol constants (handler names, error types, defaults).
contrib/temporal-workflowstreams/src/main/java/io/temporal/workflowstreams/WorkflowStreamClientOptions.java	Client-side configuration options for batching and subscription polling.
contrib/temporal-workflowstreams/src/main/java/io/temporal/workflowstreams/WorkflowStreamClient.java	External client API wrapping the publisher and subscription driver(s).
contrib/temporal-workflowstreams/src/main/java/io/temporal/workflowstreams/WorkflowStream.java	Workflow-side stream host implementing publish/poll/offset/truncate and CAN state capture.
contrib/temporal-workflowstreams/src/main/java/io/temporal/workflowstreams/WireItem.java	Wire DTO for poll results / state snapshots with fixed JSON field names.
contrib/temporal-workflowstreams/src/main/java/io/temporal/workflowstreams/TopicHandle.java	Client-side per-topic publish/subscribe convenience API.
contrib/temporal-workflowstreams/src/main/java/io/temporal/workflowstreams/SubscribeOptions.java	Subscription options (topics, start offset, poll cooldown).
contrib/temporal-workflowstreams/src/main/java/io/temporal/workflowstreams/PublishInput.java	Wire DTO for publish signal payload (items + dedup metadata).
contrib/temporal-workflowstreams/src/main/java/io/temporal/workflowstreams/PublishEntry.java	Wire DTO for a single published entry (topic + encoded payload).
contrib/temporal-workflowstreams/src/main/java/io/temporal/workflowstreams/PollResult.java	Wire DTO for poll update response (items/next_offset/more_ready).
contrib/temporal-workflowstreams/src/main/java/io/temporal/workflowstreams/PollInput.java	Wire DTO for poll update request (topics/from_offset).
contrib/temporal-workflowstreams/src/main/java/io/temporal/workflowstreams/internal/SubscriptionDriver.java	Shared long-poll engine implementing rollover/terminal/truncation semantics and backpressure.
contrib/temporal-workflowstreams/src/main/java/io/temporal/workflowstreams/internal/StreamPublisher.java	Background batching publisher with retry window and dedup sequencing.
contrib/temporal-workflowstreams/src/main/java/io/temporal/workflowstreams/internal/PayloadWire.java	Base64-of-proto payload wire encode/decode utility.
contrib/temporal-workflowstreams/src/main/java/io/temporal/workflowstreams/FlushTimeoutException.java	Exception surfaced when publish retries exceed configured max retry duration.
contrib/temporal-workflowstreams/README.md	Module documentation and usage examples for hosting/publishing/subscribing.
contrib/temporal-workflowstreams/build.gradle	Gradle configuration for the new contrib module (deps + tests).
.github/CODEOWNERS	Adds ownership entry for the new contrib module and adjusts formatting.

---

💡 Add Copilot custom instructions for smarter, more guided reviews. Learn how to get started.

[xumaple]

Here dataConverter.toPayload(value) throws DataConverterException when no configured converter accepts the value. If, say, a caller narrowed the converter set to byte-array-only and then tried to publish a String, the chain would reject. (Or alternatively consider if a default converter was given an unserializable object.)

encodeBuffer throws before buffer is reset or pending is set, so doFlush leaves the poisoned buffer intact. backgroundFlush then catches it as if it were a transient send failure (L129) and retries the same buffer every batchInterval, forever. Net effect: one unconvertible value permanently wedges the publisher, causing every subsequent (valid) item buffers behind it and never ships, and nothing surfaces to a fire-and-forget caller unless they happen to call flush().

Worth contrasting with the workflow side: WorkflowStream.publishToTopic uses the same conversion, but there a failure surfaces as a workflow-task failure (visible, retried). On the client it's swallowed and silent.

Consider treating a conversion failure as fatal-to-that-item rather than transient: drop (or quarantine) the offending entry and record it in deferredError so flush()/close() surface it, instead of re-encoding the whole buffer each tick.