brianstrauch · Oct 20, 2026
@@ -69,6 +69,7 @@ to docs, or any other relevant information.
   specific to the Temporal CLI dev server implementation and may no longer be valid if that implementation changes.
 
 ### Fixed
+- Workflow Streams subscriptions now deliver retained items after workflow completion, failure, cancellation, termination, or timeout using a query pinned to the closed run.
 - `ActivityClient` now sends the `temporal-namespace` gRPC header like `WorkflowClient` and `ScheduleClient` do,
   so standalone activity calls are no longer rejected with `PERMISSION_DENIED` by namespace-scoped auth (e.g.
   Temporal Cloud API keys).

@@ -4,7 +4,8 @@ A durable publish/subscribe log hosted inside a Temporal Workflow.
 
 External code (activities, starters, other processes) publishes messages to
 named topics via **signals**; subscribers long-poll for new items via
-**updates**; a **query** exposes the current offset. The stream is backed by
+**updates**, then read retained items after closure via **queries**; a separate
+**query** exposes the current offset. The stream is backed by
 Temporal's durable execution, giving ordered, durable, exactly-once delivery
 with client-side batching, publisher dedup, continue-as-new survival,
 truncation, and ~1 MB response paging.
@@ -18,7 +19,7 @@ All APIs in this module are experimental and may change.
 ## Workflow side
 
 Construct a `WorkflowStream` once in a `@WorkflowInit` constructor. The factory
-registers the publish signal, poll update, and offset query handlers, and a
+registers the publish signal, poll update, poll query, and offset query handlers. A
 `@WorkflowInit` constructor runs before any handler dispatch, so polls and
 offset queries arriving with the first workflow task (e.g. from
 update-with-start) are accepted rather than rejected.
@@ -114,6 +115,18 @@ when the buffer reaches the max batch size, on `forceFlush`, on an explicit
 
 ## Subscribing
 
+Subscriptions keep polling through updates while the workflow runs. After a
+confirmed terminal state, they query that run for its remaining retained items
+and finish after delivering the final page. The poll query shares
+`__temporal_workflow_stream_poll` with the update, returns immediately, and reads
+from the retained base when the requested offset was truncated. Continue-as-new
+still follows the successor run.
+
+Upgrade workers before using upgraded clients. Closed-stream reads require
+retained history and an available compatible worker; configured query rejection
+and query failures are surfaced to the subscriber. Query reads do not add history
+events.
+
 There are two subscriber APIs over one shared poll engine: a non-blocking
 listener and a blocking iterator. Neither occupies a thread while a poll is
 blocked on the server — polling runs on a small executor shared by all of a
@@ -206,7 +219,7 @@ the per-item payload encoding (base64 of the serialized
 `temporal.api.common.v1.Payload`) match other languages' packages
 exactly, so a Java publisher or subscriber interoperates with a workflow
 written in any of them and vice versa. The data converter codec chain
-(encryption, compression) runs once on the signal/update envelope — never per
+(encryption, compression) runs once on the signal/update/query envelope — never per
 item — so payloads are not double-encoded.
 
 One Java-specific caveat: the protocol envelope types are serialized by the

@@ -6,7 +6,7 @@
 import java.util.List;
 
 /**
- * The poll update payload: a request to long-poll for new items.
+ * The poll update or query payload: a request for items at a global offset.
  *
  * <p>Field names are part of the cross-language wire protocol; this type must serialize to JSON
  * with exactly these names.

@@ -6,8 +6,8 @@
 import java.util.List;
 
 /**
- * The poll update response: items matching the poll request. When {@code more_ready} is true the
- * response was truncated to stay within size limits and the subscriber should poll again
+ * The poll update or query response: items matching the poll request. When {@code more_ready} is
+ * true the response was truncated to stay within size limits and the subscriber should poll again
  * immediately rather than applying a cooldown.
  *
  * <p>Field names are part of the cross-language wire protocol; this type must serialize to JSON

@@ -20,10 +20,10 @@
 
 /**
  * The workflow-side stream object: an append-only, multi-topic log served to external publishers
- * (via signal), subscribers (via update), and offset queries (via query).
+ * (via signal) and subscribers (via update while running and query after closure).
  *
  * <p>Construct it once with {@link #newInstance}, preferably in a {@link
- * io.temporal.workflow.WorkflowInit} constructor: the factory registers all three handlers on the
+ * io.temporal.workflow.WorkflowInit} constructor: the factory registers all stream handlers on the
  * current workflow, and a {@code @WorkflowInit} constructor runs before any handler dispatch, so
  * polls arriving with the first workflow task are accepted. Constructing it at the start of the
  * workflow method also works — signals received earlier are buffered by the SDK — but polls and
@@ -274,6 +274,15 @@ public PollResult poll(PollInput input) {
             WorkflowStreamConstants.ERROR_TYPE_TRUNCATED_OFFSET);
       }
 
+      return collectBatch(input);
+    }
+
+    @Override
+    public PollResult pollQuery(PollInput input) {
+      return collectBatch(input);
+    }
+
+    private PollResult collectBatch(PollInput input) {
       long logOffset = Math.max(input.fromOffset - baseOffset, 0);
 
       Set<String> topicSet =

@@ -142,8 +142,9 @@ public long getOffset() {
    *
    * <p>The consuming thread blocks waiting for items; polling itself runs on the client's poll
    * executor. Each item carries the raw {@link io.temporal.api.common.v1.Payload}; decode it with
-   * your data converter. The subscription ends cleanly when the workflow reaches a terminal state,
-   * automatically follows continue-as-new chains, and also ends when this client is closed.
+   * your data converter. The subscription delivers the retained tail before ending when the
+   * workflow reaches a terminal state, automatically follows continue-as-new chains, and also ends
+   * when this client is closed.
    */
   public WorkflowStreamSubscription subscribe(SubscribeOptions options) {
     return new WorkflowStreamSubscription(listener -> newSubscriptionDriver(options, listener));
@@ -155,9 +156,10 @@ public WorkflowStreamSubscription subscribe(SubscribeOptions options) {
    * which is never held while a poll is blocked on the server, so many subscriptions share a small
    * pool. Delivery starts immediately.
    *
-   * <p>The stream ends cleanly with {@link WorkflowStreamListener#onCompleted} when the workflow
-   * reaches a terminal state, automatically follows continue-as-new chains, and reports
-   * unrecoverable failures to {@link WorkflowStreamListener#onError}. Stop it early with {@link
+   * <p>After delivering retained items, the stream ends with {@link
+   * WorkflowStreamListener#onCompleted} when the workflow reaches a terminal state, automatically
+   * follows continue-as-new chains, and reports unrecoverable failures to {@link
+   * WorkflowStreamListener#onError}. Stop it early with {@link
    * WorkflowStreamSubscriptionHandle#close}; closing this client also stops it.
    */
   public WorkflowStreamSubscriptionHandle subscribe(

@@ -17,6 +17,9 @@ public final class WorkflowStreamConstants {
   /** Update subscribers send to long-poll for new items. */
   public static final String POLL_UPDATE_NAME = "__temporal_workflow_stream_poll";
 
+  /** Query subscribers use to read retained items after the workflow closes. */
+  public static final String POLL_QUERY_NAME = "__temporal_workflow_stream_poll";
+
   /** Query that returns the current global offset. */
   public static final String OFFSET_QUERY_NAME = "__temporal_workflow_stream_offset";
 

@@ -22,6 +22,9 @@ public interface WorkflowStreamHandlers {
   @UpdateMethod(name = WorkflowStreamConstants.POLL_UPDATE_NAME)
   PollResult poll(PollInput input);
 
+  @QueryMethod(name = WorkflowStreamConstants.POLL_QUERY_NAME)
+  PollResult pollQuery(PollInput input);
+
   @UpdateValidatorMethod(updateName = WorkflowStreamConstants.POLL_UPDATE_NAME)
   void validatePoll(PollInput input);
 

@@ -3,6 +3,7 @@
 import io.temporal.api.enums.v1.WorkflowExecutionStatus;
 import io.temporal.client.UpdateOptions;
 import io.temporal.client.WorkflowClient;
+import io.temporal.client.WorkflowExecutionDescription;
 import io.temporal.client.WorkflowStub;
 import io.temporal.client.WorkflowTargetOptions;
 import io.temporal.client.WorkflowUpdateHandle;
@@ -63,6 +64,7 @@ public final class SubscriptionDriver implements WorkflowStreamSubscriptionHandl
   // update's outcome so that, if that run continues-as-new mid-poll (failing the outcome), we
   // still know which run to inspect to tell a rollover apart from a terminal end.
   private String polledRunId = "";
+  private WorkflowStub terminalRunStub;
 
   private volatile boolean closed;
   private final AtomicBoolean started = new AtomicBoolean();
@@ -114,6 +116,16 @@ private void poll() {
       finishSilent();
       return;
     }
+    if (terminalRunStub != null) {
+      // Query failures must surface directly, without reclassifying a closed run.
+      deliver(
+          terminalRunStub.query(
+              WorkflowStreamConstants.POLL_QUERY_NAME,
+              PollResult.class,
+              new PollInput(topics, offset)));
+      return;
+    }
+    polledRunId = "";
     WorkflowUpdateHandle<PollResult> handle;
     try {
       // Wait only for ACCEPTED so startUpdate returns the handle (and its run id) as soon
@@ -170,6 +182,8 @@ private void deliverFrom(List<WorkflowStreamItem> items, int start, boolean more
       if (i == items.size()) {
         if (moreReady) {
           hop(this::poll);
+        } else if (terminalRunStub != null) {
+          finishCompleted();
         } else {
           schedule(this::poll, pollCooldownMs);
         }
@@ -232,14 +246,21 @@ private void handleError(Throwable e) {
     // The workflow may have continued-as-new or completed between polls. Follow the chain,
     // exit cleanly on a terminal state, otherwise surface the error. describe() is a blocking
     // call, which is why error handling always runs on the executor.
-    WorkflowExecutionStatus status = describePolledRun();
+    WorkflowExecutionDescription description = describePolledRun();
+    WorkflowExecutionStatus status = description.getStatus();
     if (status == WorkflowExecutionStatus.WORKFLOW_EXECUTION_STATUS_CONTINUED_AS_NEW) {
       // Subsequent polls use the latest-run stub, addressing the successor automatically.
       hop(this::poll);
       return;
     }
-    if (isTerminal(status)) {
-      finishCompleted();
+    if (isTerminal(status) && !description.getExecution().getRunId().isEmpty()) {
+      terminalRunStub =
+          client.newUntypedWorkflowStub(
+              WorkflowTargetOptions.newBuilder()
+                  .setWorkflowId(workflowId)
+                  .setRunId(description.getExecution().getRunId())
+                  .build());
+      hop(this::poll);
       return;
     }
     finishError(e);
@@ -252,23 +273,19 @@ private void handleError(Throwable e) {
    * the latest run automatically. A blank run id (no poll has been admitted yet) falls back to
    * describing the latest run.
    */
-  private WorkflowExecutionStatus describePolledRun() {
-    try {
-      WorkflowStub stub;
-      if (polledRunId == null || polledRunId.isEmpty()) {
-        stub = latestRunStub;
-      } else {
-        stub =
-            client.newUntypedWorkflowStub(
-                WorkflowTargetOptions.newBuilder()
-                    .setWorkflowId(workflowId)
-                    .setRunId(polledRunId)
-                    .build());
-      }
-      return stub.describe().getStatus();
-    } catch (RuntimeException e) {
-      return WorkflowExecutionStatus.WORKFLOW_EXECUTION_STATUS_UNSPECIFIED;
+  private WorkflowExecutionDescription describePolledRun() {
+    WorkflowStub stub;
+    if (polledRunId == null || polledRunId.isEmpty()) {
+      stub = latestRunStub;
+    } else {
+      stub =
+          client.newUntypedWorkflowStub(
+              WorkflowTargetOptions.newBuilder()
+                  .setWorkflowId(workflowId)
+                  .setRunId(polledRunId)
+                  .build());
     }
+    return stub.describe();
   }
 
   private void finishSilent() {

@@ -2,6 +2,7 @@
 
 import io.temporal.api.common.v1.WorkflowExecution;
 import io.temporal.client.WorkflowClient;
+import io.temporal.client.WorkflowOptions;
 import io.temporal.client.WorkflowStub;
 import io.temporal.common.converter.DataConverter;
 import io.temporal.common.converter.DefaultDataConverter;
@@ -176,6 +177,66 @@ public void testTerminalCallsOnCompleted() throws Exception {
     }
   }
 
+  @Test
+  public void testClosedStreamCompletionWaitsForDelivery() throws Exception {
+    WorkflowStub stub = startHostWorkflow();
+    stub.signal("publishLocal", "evt", "tail");
+    stub.signal("finish");
+    stub.getResult(Void.class);
+    RecordingListener listener = new RecordingListener();
+    CompletableFuture<Void> gate = new CompletableFuture<>();
+    listener.gates.add(gate);
+    try (WorkflowStreamClient streamClient = newStreamClient(stub);
+        WorkflowStreamSubscriptionHandle handle = streamClient.subscribe(FAST_POLL, listener)) {
+      awaitItems(listener, 1);
+      Assert.assertEquals("tail", decode(listener.items.get(0)));
+      Assert.assertEquals(1, listener.completed.getCount());
+      Assert.assertFalse(handle.getDoneFuture().isDone());
+      gate.complete(null);
+      handle.getDoneFuture().get(TIMEOUT_MS, TimeUnit.MILLISECONDS);
+      Assert.assertEquals(0, listener.completed.getCount());
+      Assert.assertNull(listener.error.get());
+    }
+  }
+
+  @Test
+  public void testClosedStreamPagesStayPinnedAfterWorkflowIdReuse() throws Exception {
+    WorkflowStub stub = startHostWorkflow();
+    String value = new String(new char[400_000]).replace('\0', 'x');
+    for (int i = 0; i < 3; i++) {
+      stub.signal("publishLocal", "evt", value);
+    }
+    stub.signal("finish");
+    stub.getResult(Void.class);
+    RecordingListener listener = new RecordingListener();
+    CompletableFuture<Void> gate = new CompletableFuture<>();
+    listener.gates.add(gate);
+    try (WorkflowStreamClient streamClient = newStreamClient(stub);
+        WorkflowStreamSubscriptionHandle handle = streamClient.subscribe(FAST_POLL, listener)) {
+      awaitItems(listener, 1);
+      SubscribeHostWorkflow replacement =
+          testWorkflowRule
+              .getWorkflowClient()
+              .newWorkflowStub(
+                  SubscribeHostWorkflow.class,
+                  WorkflowOptions.newBuilder()
+                      .setWorkflowId(stub.getExecution().getWorkflowId())
+                      .setTaskQueue(testWorkflowRule.getTaskQueue())
+                      .build());
+      WorkflowClient.start(replacement::execute, null);
+      replacement.publishLocal("evt", "replacement");
+      gate.complete(null);
+      handle.getDoneFuture().get(TIMEOUT_MS, TimeUnit.MILLISECONDS);
+      Assert.assertEquals(3, listener.items.size());
+      for (int i = 0; i < 3; i++) {
+        Assert.assertEquals(value, decode(listener.items.get(i)));
+        Assert.assertEquals(i, listener.items.get(i).getOffset());
+      }
+      replacement.finish();
+      WorkflowStub.fromTyped(replacement).getResult(Void.class);
+    }
+  }
+
   @Test
   public void testListenerFollowsContinueAsNew() throws Exception {
     WorkflowStub stub = startHostWorkflow();

@@ -1,12 +1,14 @@
 package io.temporal.workflowstreams;
 
+import com.fasterxml.jackson.databind.ObjectMapper;
 import com.google.protobuf.ByteString;
 import io.temporal.api.common.v1.Payload;
 import io.temporal.common.converter.DataConverter;
 import io.temporal.common.converter.DefaultDataConverter;
 import io.temporal.workflowstreams.internal.PayloadWire;
 import java.nio.charset.StandardCharsets;
 import java.util.Base64;
+import java.util.Collections;
 import org.junit.Assert;
 import org.junit.Test;
 
@@ -65,6 +67,32 @@ public void testBinaryPayloadRoundTrip() {
     Assert.assertArrayEquals(original, b);
   }
 
+  @Test
+  public void testSharedClosedStreamPollVector() throws Exception {
+    String wire = "ChgKCGVuY29kaW5nEgxiaW5hcnkvcGxhaW4SA2FiYw==";
+    Payload payload =
+        Payload.newBuilder()
+            .putMetadata("encoding", ByteString.copyFromUtf8("binary/plain"))
+            .setData(ByteString.copyFromUtf8("abc"))
+            .build();
+    Assert.assertEquals(wire, PayloadWire.encode(payload));
+    Assert.assertEquals(payload, PayloadWire.decode(wire));
+    ObjectMapper mapper = new ObjectMapper();
+    Assert.assertEquals(
+        mapper.readTree("{\"topics\":[\"events\"],\"from_offset\":3}"),
+        mapper.readTree(
+            mapper.writeValueAsString(new PollInput(Collections.singletonList("events"), 3))));
+    Assert.assertEquals(
+        mapper.readTree(
+            "{\"items\":[{\"topic\":\"events\",\"data\":\""
+                + wire
+                + "\",\"offset\":3}],\"next_offset\":4,\"more_ready\":false}"),
+        mapper.readTree(
+            mapper.writeValueAsString(
+                new PollResult(
+                    Collections.singletonList(new WireItem("events", wire, 3)), 4, false))));
+  }
+
   @Test
   public void testWireSize() {
     Payload payload =