observability/streaming

110 notes in this subject, read out of this brain and free to use. This is page 2 of 2.

WorkflowStreamState field type requirement

The stream_state field type must include | None: stream_state: WorkflowStreamState | None = None. Always use the concrete type, not Any. With Any, the data converter rebuilds the field as a plain dict and WorkflowStream(prior_state=...) raises an AttributeError accessing .log / .base_offset / .publishers on the dict.

Continue-as-new with stream state example

from dataclasses import dataclass, field from temporalio import workflow from temporalio.contrib.workflow_streams import WorkflowStream, WorkflowStreamState @dataclass class AppState: items_processed: int = 0 @dataclass class WorkflowInput: app_state: AppState = field(default_factory=AppState) stream_state: WorkflowStreamState | None = None @workflow.defn class LongRunningWorkflow: @workflow.init def __init__(self, input: WorkflowInput) -> None: self.app_state = input.app_state self.stream = WorkflowStream(prior_state=input.stream_state) @workflow.run async def run(self, input: WorkflowInput) -> None: while True: await do_one_iteration(self) if workflow.info().is_continue_as_new_suggested(): await self.stream.continue_as_new( lambda stream_state: [ WorkflowInput( app_state=self.app_state, stream_state=stream_state, ) ] )

Pass additional continue_as_new parameters with explicit recipe

To pass other Continue-As-New parameters such as task_queue, retry_policy, run_timeout, use the explicit recipe: call self.stream.detach_pollers(), await workflow.wait_condition(workflow.all_handlers_finished), then call workflow.continue_as_new() with the desired parameters and stream state via self.stream.get_state().

publisher_ttl for deduplication window

publisher_ttl is a limit on the deduplication window. At each Continue-As-New, deduplicate entries whose last_seen is older than this are dropped. last_seen is updated on each successful publish, so a publisher that retries through a long partition without success can still time out. Tune upward if publishers can be silent for extended windows by passing publisher_ttl= to WorkflowStream.continue_as_new().

WorkflowStreamClient is asyncio-only

WorkflowStreamClient is asyncio-only. The client buffer is mutated on the publish path and read from the flusher inside a single event loop. Do not call publish() from a Worker thread.

Custom handlers and stream state race condition

Custom handlers read stream state on the first activation. WorkflowStream registers its publish-Signal handler dynamically from __init__, so on the first activation a publish Signal can be queued before class-level @workflow.signal or @workflow.update handlers have run. A handler that observes state set by stream initialization in that same activation can see pre-publish state. The fix is to make the handler async def and await once before reading state. asyncio.sleep(0) is a no-op yield that suffices and adds no history events. Don't substitute workflow.sleep(0), which records a timer event.

LLM streaming activity example

from openai import AsyncOpenAI from dataclasses import dataclass @dataclass class TextDelta: text: str @activity.defn async def stream_completion(prompt: str) -> str: stream_client = WorkflowStreamClient.from_within_activity( batch_interval=timedelta(milliseconds=200), ) openai_client = AsyncOpenAI(max_retries=0) async with stream_client: deltas = stream_client.topic("delta", type=TextDelta) retry = stream_client.topic("retry", type=dict) close = stream_client.topic("close") if activity.info().attempt > 1: retry.publish({"attempt": activity.info().attempt}, force_flush=True) full: list[str] = [] first = True oai_stream = await openai_client.chat.completions.create( model="gpt-4o-mini", messages=[{"role": "user", "content": prompt}], stream=True, ) async for chunk in oai_stream: if not chunk.choices: continue text = chunk.choices[0].delta.content if not text: continue deltas.publish(TextDelta(text=text), force_flush=first) first = False full.append(text) close.publish({}) return "".join(full)

LLM streaming workflow example

@workflow.defn class ChatWorkflow: @workflow.init def __init__(self, input: ChatInput) -> None: self.stream = WorkflowStream() self.subscriber_done: bool = False @workflow.signal async def subscriber_acknowledged_terminator(self) -> None: self.subscriber_done = True @workflow.run async def run(self, input: ChatInput) -> str: result = await workflow.execute_activity( stream_completion, input.prompt, start_to_close_timeout=timedelta(minutes=5), ) try: await workflow.wait_condition( lambda: self.subscriber_done, timeout=timedelta(seconds=30), ) except TimeoutError: pass return result

LLM streaming consumer example

async def stream_chat(chat_id: str) -> str: stream = WorkflowStreamClient.create(temporal_client, workflow_id=chat_id) converter = temporal_client.data_converter.payload_converter output: list[str] = [] def render() -> None: ... # display the accumulated output async for item in stream.subscribe( ["delta", "retry", "close"], result_type=RawValue ): if item.topic == "retry": output.clear() render() elif item.topic == "delta": delta = converter.from_payload(item.data.payload, TextDelta) output.append(delta.text) render() elif item.topic == "close": await temporal_client.get_workflow_handle(chat_id).signal( ChatWorkflow.subscriber_acknowledged_terminator ) break return "".join(output)

UI display of summaries in Event History

In the Temporal UI Event History, individual events display their associated summaries in purple text next to their corresponding events, providing immediate context without requiring event expansion. When an event is expanded, the summary is also prominently displayed in the detailed view.

UI Workflow Overview section displays workflow metadata

The Temporal UI Workflow Overview section at the top of the workflow details page displays workflow-level metadata, including Summary & Details (showing static summary and details set when starting the workflow) and Current Details (showing dynamic details that can be updated during execution). All workflow details support standard Markdown formatting excluding images, HTML, and scripts.

UI display of activity and timer summaries on Timeline

In the Temporal UI Timeline tab on a Workflow details page, Activity and Timer summaries are displayed as labels directly on the horizontal bars representing each activity or timer instance. Labels longer than 120 characters are truncated with an ellipsis. This feature is especially useful for fan-out workflows that schedule many instances of the same Activity Type. Activity Summary support on the Timeline shipped in Temporal UI v2.34.6 and is available on Temporal Cloud and self-hosted UI builds at that version or later.

Publish from Activity within Activity context

Inside an Activity scheduled by a Workflow, use WorkflowStreamClient.fromWithinActivity() to infer the Temporal Client and the parent Workflow Id from the Activity context, so you don't have to thread them through the Activity's input: await using client = WorkflowStreamClient.fromWithinActivity(); const deltas = client.topic<Delta>('delta');

Publish from Activity using WorkflowStreamClient

When events originate in an Activity, publish from the Activity directly rather than returning them for the Workflow to forward. The Workflow hosts the stream but doesn't read its own stream; it processes the Activity's return value and emits its own lifecycle events. Keeping Workflow state independent of streamed output lets retried Activity attempts surface to subscribers without polluting the Workflow's durable state.

Publish from a client using WorkflowStreamClient

Any process that has a Temporal Client and the target Workflow Id can publish to that Workflow's stream by constructing a WorkflowStreamClient with: WorkflowStreamClient.create(client, workflowId). Then use it the same way as the Workflow-side handle: bind a topic, publish through it, and let await using flush on scope exit.

Type binding enforcement in TypeScript Workflow Streams

In TypeScript, the type parameter T on stream.topic<T>(name) is a compile-time annotation only because TypeScript has no runtime type representation. The library cannot enforce per-topic type uniformity at the publish site. If two publishers bind the same topic name to different types, the mismatch is not caught at publish. The subscriber gets a decode error when it processes events from the mismatched publisher. A pre-built Payload may be passed to publish() regardless of the handle's type T, taking the zero-copy fast path.

Payload conversion in publish()

publish() runs the default payload converter to encode each value. The codec chain (encryption, compression, etc.) runs once on the Signal or Update envelope that carries the batch, never per item, so encryption and compression are applied exactly once in each direction.

Topic binding and publish pattern

Bind a topic name to its event type once via stream.topic<T>(name), then call publish() on the returned handle to append events. The handle carries the topic name and the type T so call sites don't have to repeat them on every publish. Repeated calls with the same name return the same handle instance. This pattern ensures subscribers reading the same handle decode to the matching type.

Publish from a Workflow example

import { WorkflowStream } from '@temporalio/workflow-streams/workflow'; export interface StatusEvent { state: string; progress?: number; detail?: string; } export interface OrderInput { orderId: string; } export async function orderWorkflow(input: OrderInput): Promise<void> { const stream = new WorkflowStream(); const status = stream.topic<StatusEvent>('status'); status.publish({ state: 'validating', detail: 'checking inventory' }); await validateOrder(input.orderId); status.publish({ state: 'charging', progress: 33, detail: 'authorizing payment' }); await chargePayment(input.orderId); status.publish({ state: 'shipping', progress: 66, detail: 'dispatching to warehouse' }); await dispatchOrder(input.orderId); status.publish({ state: 'completed', progress: 100 }); }

Enable streaming on a Workflow

Create exactly one WorkflowStream instance at the top of the Workflow function, before any await. Construction must happen at the top because the stream's handlers (publish Signal, subscribe Update, and offset Query handlers) have to be registered before the first publish Signal arrives. Doing it after an await would miss any publishes that arrived before the run body resumed. If more than one WorkflowStream is constructed on the same Workflow, the handlers are silently replaced because the TypeScript Workflow runtime doesn't expose an inspection API for existing handlers.

Workflow Streams library imports

The Workflow Streams library ships as @temporalio/workflow-streams and provides two subpaths for imports. Import from @temporalio/workflow-streams/workflow for the Workflow-safe interface (WorkflowStream, WorkflowStreamState, etc) which bundles cleanly into Workflow code. Import from @temporalio/workflow-streams/client for the client interface (WorkflowStreamClient, etc) which pulls in crypto, @temporalio/activity, and @temporalio/client—none of which resolve inside the Workflow sandbox. Do not import from a Workflow file.

Multiple WorkflowStreams silently replace handlers

Constructing two WorkflowStreams silently replaces handlers. The TypeScript Workflow runtime doesn't expose an inspection API for existing handlers, so the library can't raise on a duplicate the way the Python SDK does. Construct exactly one WorkflowStream per Workflow at the top of the function.

Type bindings not shared across publishers

Type bindings aren't shared across publishers. Each WorkflowStream and each WorkflowStreamClient records topic types only for its own instance, and the type parameter T is erased at compile time, so no runtime check enforces uniformity. If two publishers bind the same topic name to different types, the mismatch is not caught at publish, and the subscriber gets a decode error when it processes events from the mismatched publisher.

Custom payload converters with Workflow Streams

A WorkflowStreamClient created via WorkflowStreamClient.create(client, ...) picks up the client's configured payload converter. Subscribers decode through the same converter. The Workflow-side always uses defaultPayloadConverter. If you ship a custom converter, make sure both sides agree or use types the default converter handles.

Cross-realm Uint8Array for binary publishes

Hand-publishing a Uint8Array from Workflow code uses a dedicated code path that constructs a binary/plain Payload directly, because the sandbox's TextEncoder returns a host-realm Uint8Array that fails instanceof checks against the sandbox's own globals. Generally this doesn't need to be thought about, but if you bypass the workflow-side handle and construct payloads manually, use the Workflow-side WorkflowStream API rather than building payloads by hand.

Stream LLM output with retry handling

When an Activity can retry, the consumer side has to account for it. A retried attempt is a fresh publisher, so its output appears in the stream alongside the output from the previous attempt. In the LLM streaming pattern, that means the failed attempt's partial deltas and the retried attempt's full output both reach a subscribed UI unless the UI resets on a RETRY event. The Activity publishes a RETRY event when Context.current().info.attempt > 1. This lets the UI respond appropriately to the failure, typically by clearing accumulated deltas before the next attempt's deltas arrive.

Activity streaming with first delta forceFlush

import { Context } from '@temporalio/activity'; import { WorkflowStreamClient } from '@temporalio/workflow-streams/client'; import OpenAI from 'openai'; export interface TextDelta { text: string; } export interface RetryEvent { attempt: number; } export async function streamCompletion(prompt: string): Promise<string> { const attempt = Context.current().info.attempt; await using streamClient = WorkflowStreamClient.fromWithinActivity({ batchInterval: '200 milliseconds', }); const openai = new OpenAI({ maxRetries: 0 }); const deltas = streamClient.topic<TextDelta>('delta'); const retry = streamClient.topic<RetryEvent>('retry'); const close = streamClient.topic<Record<string, never>>('close'); if (attempt > 1) { retry.publish({ attempt }, { forceFlush: true }); } const oaiStream = await openai.chat.completions.create({ model: 'gpt-4o-mini', messages: [{ role: 'user', content: prompt }], stream: true, }); const full: string[] = []; let first = true; for await (const chunk of oaiStream) { const text = chunk.choices[0]?.delta?.content; if (!text) continue; deltas.publish({ text }, first ? { forceFlush: true } : undefined); first = false; full.push(text); } close.publish({}); return full.join(''); }

Workflow with stream acknowledgment handshake

import { condition, defineSignal, executeActivity, setHandler } from '@temporalio/workflow'; import { WorkflowStream } from '@temporalio/workflow-streams/workflow'; import type * as activities from './activities'; export const subscriberAcknowledgedTerminator = defineSignal('subscriberAcknowledgedTerminator'); export interface ChatInput { prompt: string; } export async function chatWorkflow(input: ChatInput): Promise<string> { const stream = new WorkflowStream(); let subscriberDone = false; setHandler(subscriberAcknowledgedTerminator, () => { subscriberDone = true; }); const result = await executeActivity<typeof activities.streamCompletion>('streamCompletion', input.prompt, { startToCloseTimeout: '5 minutes', }); await condition(() => subscriberDone, '30 seconds'); return result; }

Consumer stream subscription with heterogeneous topics

import { Client } from '@temporalio/client'; import { defaultPayloadConverter } from '@temporalio/common'; import { WorkflowStreamClient } from '@temporalio/workflow-streams/client'; import { subscriberAcknowledgedTerminator } from './workflows'; export async function streamChat(chatId: string): Promise<string> { const temporalClient = new Client(); const stream = WorkflowStreamClient.create(temporalClient, chatId); const output: string[] = []; function render(): void { // ... display the accumulated output (terminal redraw, UI update, etc.) } for await (const item of stream.subscribe(['delta', 'retry', 'close'])) { if (item.topic === 'retry') { output.length = 0; render(); } else if (item.topic === 'delta') { const delta = defaultPayloadConverter.fromPayload<TextDelta>(item.data); output.push(delta.text); render(); } else if (item.topic === 'close') { await temporalClient.workflow.getHandle(chatId).signal(subscriberAcknowledgedTerminator); break; } } return output.join(''); }

WorkflowStreamClient single event-loop constraint

WorkflowStreamClient is single-event-loop. The client buffer is mutated on the publish path and read from the background flusher inside one Node event loop. Do not call publish() from a Worker thread. Route events back to the loop that owns the client.

Deduplication window maxRetryDuration limit

maxRetryDuration: A WorkflowStreamClient retries a failed batch for up to this long. If the duration elapses with the batch still pending, the client gives up, the pending batch is dropped, and a FlushTimeoutError is raised. On timeout, the dropped batch is at-most-once: it may or may not have reached the Workflow. Subsequent publishes resume cleanly with the next sequence. One operational caveat: the FlushTimeoutError is raised from inside the background flusher task and terminates it. Until you call await client.flush() or exit the await using scope, subsequent publishes accumulate in the buffer with no flusher to ship them.

Deduplication window publisherTtl limit

publisherTtl: At each Continue-As-New, deduplicate entries whose lastSeen is older than this are dropped. lastSeen is updated on each successful publish (not on each retry attempt), so a publisher that retries through a long partition without success can still age out. A publisher that returns after a longer pause may produce a duplicate. Tune upward if your publishers can be silent for extended windows: stream.continueAsNew(buildArgs, { publisherTtl: '...' })

Stream state size consideration

The carried WorkflowStreamState includes the entire in-memory log of the previous run, so streams that carry large items can hit Temporal's per-payload size limit at the rollover. Offload the bytes via External Storage so each item is a small reference rather than the full payload, and combine that with truncate() to keep the carried log itself small.

Stream Continue-As-New with additional parameters

To pass other Continue-As-New parameters such as taskQueue, searchAttributes, or workflowRunTimeout, use the explicit recipe with makeContinueAsNewFunc instead: import { allHandlersFinished, condition, makeContinueAsNewFunc } from '@temporalio/workflow'; stream.detachPollers(); await condition(allHandlersFinished); const continueWithOptions = makeContinueAsNewFunc<typeof longRunningWorkflow>({ taskQueue: 'other-tq', }); await continueWithOptions({ itemsProcessed, streamState: stream.getState(), });

Long-running workflow with stream state example

import { workflowInfo } from '@temporalio/workflow'; import { WorkflowStream, type WorkflowStreamState } from '@temporalio/workflow-streams/workflow'; export interface WorkflowInput { itemsProcessed: number; streamState?: WorkflowStreamState; } export async function longRunningWorkflow(input: WorkflowInput): Promise<void> { const stream = new WorkflowStream(input.streamState); let itemsProcessed = input.itemsProcessed; while (true) { await doOneIteration(stream); itemsProcessed++; if (workflowInfo().continueAsNewSuggested) { await stream.continueAsNew<typeof longRunningWorkflow>((state) => [ { itemsProcessed, streamState: state, }, ]); } } }

Stream from long-running Workflows with Continue-As-New

Workflows that run for hours or accumulate thousands of events need to periodically roll over via Continue-As-New to keep history bounded. Subscribers automatically follow these rollovers, but the client retained from WorkflowStreamClient.create() or fromWithinActivity() is required (clients constructed directly from a single WorkflowHandle can't re-target the new run). To keep a stream running across rollovers without subscribers seeing a gap, carry both your application state and the stream state across the boundary. Add an optional streamState?: WorkflowStreamState field to your Workflow input, pass it to the constructor, and call stream.continueAsNew(buildArgs) to invoke the rollover.

Inspect terminal workflow status after stream

subscribe() exits cleanly when the Workflow reaches COMPLETED, FAILED, CANCELLED, TERMINATED, or TIMED_OUT, but doesn't distinguish among them. If your application needs to know which state (to display success or failure to the user, log the outcome, or decide whether to retry), call await temporalClient.workflow.getHandle(workflowId).describe() after the loop returns to inspect the Workflow's status.

Close stream with acknowledgment handshake

The subscriber sends a Signal once it has the terminator; the Workflow waits up to a timeout, returning as soon as the ack arrives: import { condition, defineSignal, setHandler } from '@temporalio/workflow'; import { WorkflowStream } from '@temporalio/workflow-streams/workflow'; export const subscriberAcknowledgedTerminator = defineSignal('subscriberAcknowledgedTerminator'); export async function chatWorkflow(input: ChatInput): Promise<string> { const stream = new WorkflowStream(); let subscriberDone = false; setHandler(subscriberAcknowledgedTerminator, () => { subscriberDone = true; }); // ... do work and publish events ... await condition(() => subscriberDone, '30 seconds'); // Returns true if the ack arrived, false on timeout. Either way, fall through. return result; }

Close stream with fixed sleep

Sleep between the terminator and the return so any in-flight poll has time to fetch the terminator before the Workflow exits: import { sleep } from '@temporalio/workflow'; // at the end of the workflow function status.publish({ state: 'completed', progress: 100 }); await sleep('30 seconds'); return result;

Closing a stream with terminator

A subscriber's for await doesn't know when the publisher is done. A common pattern combines two pieces: (1) an in-band terminator—the Workflow or its Activity publishes a sentinel event the subscriber recognizes and breaks on, and (2) a brief overlap before the Workflow returns. A poll Update that is still in flight when the Workflow returns is surfaced to the iterator and consumed silently, and no new polls can complete after that. If the Workflow returns immediately after publishing the terminator, subscribers may miss it.

Heterogeneous topics subscription example

import { defaultPayloadConverter } from '@temporalio/common'; for await (const item of stream.subscribe(['status', 'progress'])) { if (item.topic === 'status') { const evt = defaultPayloadConverter.fromPayload<StatusEvent>(item.data); console.log(`[status] ${evt.state}: ${evt.detail ?? ''}`); } else if (item.topic === 'progress') { const evt = defaultPayloadConverter.fromPayload<ProgressEvent>(item.data); console.log(`[progress] ${evt.message}`); } }

Subscribe to heterogeneous topics

To consume multiple topics whose payload types differ, call client.subscribe() directly with a list of names (or subscribe() with no arguments for every topic). The default overload yields WorkflowStreamItem<Payload>, so each item arrives as the raw Payload carrying encoding metadata. Dispatch on item.topic and decode the payload with defaultPayloadConverter.fromPayload<T>(item.data).

Subscribe edge cases

Two edge cases are worth knowing: an RPC timeout where Continue-As-New can't be followed ends the iterator silently, and a validator rejection during a Continue-As-New handoff can surface as a WorkflowUpdateFailedError.

Iterator handles re-polling and pagination

The iterator returned by subscribe() handles re-polling, pagination when a poll response hits the ~1 MB cap, and Workflow-side log truncation transparently. Callers don't need to wrap the iterator for the common cases.

Subscribe example

import { Client } from '@temporalio/client'; import { WorkflowStreamClient } from '@temporalio/workflow-streams/client'; export async function watchOrder(orderId: string): Promise<void> { const temporalClient = new Client(); const stream = WorkflowStreamClient.create(temporalClient, orderId); const status = stream.topic<StatusEvent>('status'); for await (const item of status.subscribe()) { const evt = item.data; console.log(`[${(evt.progress ?? 0).toString().padStart(3)}%] ${evt.state}: ${evt.detail ?? ''}`); if (evt.state === 'completed') break; } }

Subscribe to a workflow stream

Subscribing uses the same client construction as publishing: WorkflowStreamClient.create(client, workflowId) from any process that has a Temporal Client or fromWithinActivity() inside an Activity. Once you have a client, iterate a topic handle's subscribe(), the counterpart to publish(). The handle's bound type drives decoding, so each item.data arrives as T via the client's payload converter.

Client flush for mid-stream barrier

Use await client.flush() when you need a mid-stream barrier. Successful completion of the flush is proof that the Temporal server has received all prior publications, so subsequent work that depends on those events being durable can proceed. The client stays open for further publishing afterward. Exiting await using already flushes on its way out, so the explicit call is only for barriers in the middle.

Publish is non-blocking with no backpressure

publish() is non-blocking and applies no backpressure. From an Activity or other client, it appends to the client's in-memory buffer and returns. From inside a Workflow, it appends synchronously to the in-memory log (no buffer, nothing to flush). Subscribers pull from the Workflow's log on their own schedule, so a slow subscriber doesn't slow down publishers. If a publisher emits faster than batches can ship to the server, the buffer grows: the process uses more memory, the stream falls further behind real time, and at the limit Signals can't keep up.

Publish with forceFlush for latency

Pass { forceFlush: true } on a publish to wake the background flusher so the current buffer ships without waiting for the next interval. Use it for latency-sensitive events: the first delta of a response so the user sees something fast, or punctuated events like RETRY and STATUS_CHANGE. The call returns immediately after appending to the buffer and signaling the flusher. It doesn't wait for delivery to the Workflow or to subscribers.

Standalone Activity publishing limitation

For a standalone Activity (one started directly via Client.activity.start rather than from a Workflow), there is no parent Workflow context to infer, so fromWithinActivity() throws. Fall back to the general pattern with WorkflowStreamClient.create(client, workflowId) and the target Workflow Id threaded through the Activity's input.

Give your agent this brain