Skip to main content

Stream

Stream is a real-time, resumable message delivery channel with best-effort storage. A Step or Client writes messages immediately. A Client reads them in order with a resume token.

Use a Stream for nice-to-have data that improves the user experience but is not business-critical:

  • LLM reasoning summaries, generated-text chunks, and agent progress
  • progress bars, diagnostic events, and live previews
  • transient status shown while a Step is still running

A Stream provides best-effort durability only, using an in-memory store such as Redis. Messages can be lost when trimming runs after the capacity threshold is reached or when the storage backend restarts. Use other primitives such as Channel and Attribute for guaranteed durability.

Define and register a Stream​

A Stream has a stable name, a message type, and streamCapacityBytes. streamCapacityBytes is an approximate capacity shared by all Flow executions with the same Flow type and Stream name. It is not a per-Flow-ID limit.

This example uses a buffered text writer. Dex flushes the accumulated progress before the Step result. The handler does not wait for a Stream Store acknowledgement.

class RenderPreview(Step[str]):
    def __init__(self, progress: Stream[str]) -> None:
        self.progress = progress

    async def execute(self, context: AsyncContext, input: str) -> StepDecision:
        progress = self.progress.buffered_text(context)
        progress.write(f"Rendering preview for {input}")
        progress.write(f"Preview ready for {input}")
        return graceful_complete(f"Rendered {input}")


class StreamFlow(Flow[str]):
    progress = Stream("Progress", str, 10 * 1024 * 1024)

    def __init__(self) -> None:
        self.render_preview = RenderPreview(self.progress)

    def get_steps(self) -> StepList[str]:
        return StepList.start_step(self.render_preview)

    def get_persistence_schema(self) -> PersistenceSchema:
        return PersistenceSchema.of(self.progress)

Example: examples/python/dex_examples/primitives/stream/stream_flow.py

Semantics​

Write semantics​

A direct Step write has these semantics:

  • Dex sends it immediately during WaitFor or Execute. It is not buffered until the Step completes.
  • A later Step failure does not roll the message back.
  • One method invocation can write any number of messages to the same or different Streams. Each write is an implicit heartbeat and preserves the last explicit heartbeat value.
  • The SDK only confirms local encoding and delivery to the Worker output stream. A disabled, unavailable, full, or rejecting Stream Store drops the message without failing the Step.
  • Every Step message has source #StepExecutionID. Attempts and messages from the same Step execution share that source. Source is informational, not unique, and does not deduplicate retries.

A Client can write even when the Flow instance does not exist or is no longer active. Supply a non-empty source for each write. A source may be repeated and may contain #; every call appends another message. RPC handlers cannot write through their Context, but they can send Stream messages through an injected Client dependency.

Python coroutine Steps call Stream.write without await. Python synchronous Steps must yield the returned StepOutput from a generator. Other SDKs send the frame through their Step Context and return after local enqueueing.

Buffer text chunks​

Use the buffered text writer when many small chunks form one text value, such as LLM deltas. The default flush interval is one second and the soft threshold is 16 KiB of UTF-8 data. A timer, the size threshold, or invocation completion emits the current non-empty batch. The helper preserves text exactly, never splits a chunk, and ignores empty chunks.

Go, Java, Python async, TypeScript, and Rust stop the timer and flush the tail before sending the final result or error. Python sync generators use cooperative elapsed-time checks and must yield from an explicit final flush. An empty buffer does not produce a Stream message or implicit heartbeat. Retry does not restore unsent text, and batches sent before a retry may appear again.

Resumable reads​

ReadStream returns one message with its value, resume token, creation time, and source.

  • An empty token starts at the current retained head.
  • A token older than the retained head also starts at that head.
  • Pass the returned token unchanged to read the next message.
  • When no next message exists, the call long-polls until a message arrives or the wait expires.

Resume is best effort. A slow reader can miss messages trimmed before its next read.

Listing messages​

Use ListStreamMessages to inspect retained messages from newest to oldest. The call is non-blocking. It returns immediately with up to the requested page size.

An empty before-page token starts at the retained tail. Pass the returned next-page token unchanged to read the next older page. The token is exclusive and bound to the Flow type, Flow ID, and Stream name. An empty next-page token marks the final page.

        page = await app_state.client.list_stream_messages(
            required_query("workflowId"),
            app_state.stream.progress,
            required_int_query("pageSize"),
            optional_query("beforePageToken", ""),
        )

Example: examples/python/dex_examples/primitives/stream/controller.py

Listing is a best-effort retained-message snapshot, not a transactional snapshot. New messages appended after the first page do not enter that older-page chain, but trimming can remove messages between calls. If trimming moves the retained head past the page anchor, Dex returns an empty page instead of restarting from the tail.

The server requires a positive page size and caps it with maxReadMessages. The default limit is 1000.

Use ReadStream for forward, one-message-at-a-time consumption that can long-poll and resume. Use ListStreamMessages for reverse, non-blocking pagination over currently retained messages.

Approximate capacity​

streamCapacityBytes is an approximate capacity shared by all Flow executions with the same Flow type and Stream name. It is not a per-Flow-ID limit.

The Dex server calculates the estimated size from each message's serialized value, Flow ID, source, and configured per-message overhead.

When usage reaches the trim trigger threshold, one background trimmer removes the oldest messages across all instances until usage reaches the trim target threshold. In rare cases, if trimming is too slow and usage reaches the capacity limit (streamCapacityBytes), the write request is rejected.

The server also limits the size of each Stream message. The default is 100 KiB.

Best-effort storage​

The Stream Store can use Redis for multi-server deployments or memory for a single local server. Memory contents disappear when that process restarts. Redis survives an application restart, but trimming, Redis loss, disabled storage, or capacity pressure can still remove or reject messages.