Data primitive semantics

View as MarkdownView as Markdown

This page holds the detailed rules for Attributes, RPCs, Channels, Streams, and SubFlows: parameters, defaults, limits, errors, and edge cases. The concept pages explain what each primitive is and when to use it: Attribute, RPC, Channel, Stream, Timer, and SubFlow. The SDK doc comments describe each API in its own language.

Attributes

Initial values and later writes

StartFlowOptions can set initial Attributes when the Flow is created. Initial values and Flow creation are one operation, so the first Step sees them immediately. See Attributes in StartFlowOptions.

Applications use Flow RPCs to write Attributes after a Flow starts. A Client can also use WaitForAttributeMatch to wait until a singleton Attribute or one AttributeMap instance satisfies a scalar comparison. The call returns the matched current value. See Wait for an Attribute match.

Read Attributes from outside a Flow

Application code outside the Flow reads Attribute values with the Client's GetAttributes call. One call reads any mix of singleton Attributes and AttributeMap instances.

The read has these semantics:

• One commit. The Client sends every read in one server request. All values come from the same committed state.
• Current or latest run. The call reads the current run, or the latest run when no run is open. A closed run stays readable until Dex purges it.
• No Worker. The server answers the read itself. It dispatches nothing to a Worker, uses no Worker capacity, and works while no Worker is running.
• Committed state only. The call does not see writes that a running Step or RPC has made but not yet committed. An RPC handler, by contrast, also sees the writes it made earlier in the same invocation.
• No locks. GetAttributes takes no Attribute locks. Use an RPC for a read-modify-write.
• Large and hard-delete values. The server resolves hard-delete values before it responds, and a deleted value reads as unset. The Client loads other large values from blob storage, then decodes every value with its definition. The call returns values only after all of them decode.

An RPC remains the right choice when the read needs Flow-side validation, authorization, a derived view, or Attribute locks. See Client requests: Read Attributes for the example, the unset value in each SDK, and the errors.

Indexes

An indexed Attribute can be used in SearchFlows queries to find the Flow executions that match the query.

Index type	Value
KEYWORD	One exact string.
FULL_TEXT	Text for tokenized search.
KEYWORD_ARRAY	A list of exact strings.
INT, DOUBLE, BOOL	The corresponding scalar value.
DATETIME	A date and time value.

An index key is the key in the search schema and the key you use in a query. Index keys consume a finite, namespace-level Search Attribute pool and are shared across Flow types.

The SDK defaults a regular Attribute's index key to its Attribute name. For application indexes, always override that default with an explicit generic numbered key. Keep business names on Attributes and share one typed slot pool across every Flow type in the namespace. Two logical indexed fields in the same Flow need different slots. Reuse a slot across Flow types only with the same index type; its business meaning can differ. Dex system indexes keep their reserved names.

Index type	Application index keys
KEYWORD	keyword1, keyword2, …
KEYWORD_ARRAY	keywordList1, keywordList2, …
FULL_TEXT	text1, text2, …
INT	int1, int2, …
BOOL	bool1, bool2, …
DOUBLE	double1, double2, …
DATETIME	datetime1, datetime2, …

Renaming an index does not backfill existing Flow executions. Validate changed example schemas in a fresh local dexcli dev instance. Plan any existing deployment migration separately.

A generic key can have a different business meaning in each Flow type. Application run queries must constrain FlowType. Keep the query unchanged across pages. Dex Web already scopes searches to the selected Flow type. Parenthesize caller filters before combining them with the application scope so an OR cannot bypass it. Do not index sensitive data or PII. The customer-email value in the Customer Refund example is synthetic data only.

status = Attribute(
    "primitive-attribute-status",
    str,
    index=AttributeIndex(IndexType.KEYWORD, "keyword1"),
)

Example: examples/python/dex_examples/primitives/attribute/attribute_flow.py

For example, the status Attribute above uses keyword1 as its index key. To find completed AttributeFlow executions, use this query:

FlowType = 'AttributeFlow' AND keyword1 = 'completed'

Pass the query to SearchFlows:

page = await app_state.client.search_flows(
    "FlowType = 'AttributeFlow' AND keyword1 = 'completed'",
    20,
    "",
)

Example: examples/python/dex_examples/primitives/client-apis/controller.py

Indexed AttributeMap

An indexed AttributeMap must use an explicit index key. It is not an index over all Map instances. Each instance write updates the same search field, so one instance can replace the value written by another instance. The Map instance key is not searchable.

This does not turn the Map into a KEYWORD_ARRAY. KEYWORD_ARRAY indexes the string array from one write; Dex does not combine the values from all Map instances. After the example below writes the payment instance with value authorized, you can query the current search value with:

FlowType = 'AttributeFlow' AND keyword2 = 'authorized'

If another Map instance later writes a different value, that value replaces the search value for keyword2. Use a regular Attribute with KEYWORD_ARRAY when you need to search a collection of values. Do not use an AttributeMap index to find every instance with a value.

progress = AttributeMap(
    "primitive-attribute-progress",
    str,
    index=AttributeIndex(IndexType.KEYWORD, "keyword2"),
)

Example: examples/python/dex_examples/primitives/attribute/attribute_flow.py

Hard-delete Attributes

Dex stores every value of a hard-delete Attribute only in object storage, whatever the value size. Flow history keeps only a reference. Deleting the Attribute, or setting it to null, permanently deletes the stored value. After that, neither Dex APIs nor Flow history can recover it.

Step and RPC handlers receive the value directly. The SDKs never write it to their local blob cache, and Dex Server never writes it to its blob cache.

Enable it on the definition with HardDeleteOnNull in Go, hardDeleteOnNull in Java and TypeScript, and hard_delete_on_null in Python and Rust. The option works for both Attributes and AttributeMaps.

class AttributeFlow(Flow[str]):
    model_api_key = Attribute(
        "primitive-attribute-model-api-key",
        str,
        hard_delete_on_null=True,
    )


class AttributeStep(Step[str]):
    def execute(self, context: Context, input: str) -> StepDecision:
        self.flow.status.set(context, "completed")
        # The Flow no longer needs the short-lived key, so delete it permanently.
        self.flow.model_api_key.delete(context)
        return graceful_complete(input)

Example: examples/python/dex_examples/primitives/attribute/attribute_flow.py

A hard-delete Attribute has these requirements:

• Only string and object values are supported, including bytes. Numeric and Boolean value types are rejected.
• It cannot be indexed or synced to an Attribute Store. WaitForAttributeMatch and Action conditions reject it.
• A server that supports hard-delete Attributes, such as server-zd. The SDKs refuse to write the Attribute to an older server.

Each Flow keeps one stored value per Attribute key and overwrites it in place. Plan for these consequences:

• A Step or RPC attempt writes or deletes the stored value before its result is committed. A retried attempt, other Steps and RPCs, and Attribute reads can observe a write or deletion made by an attempt that later failed. Make handlers safe to repeat instead of relying on a read-modify-write of the Attribute.
• When concurrent Steps or RPCs write the same key, the final value may not be the last committed write.
• A Flow start or SubFlow start that fails because the Flow ID is already running still overwrites the running Flow's value. A null initial value deletes it.
• Flows that reuse a Flow ID share the stored value.
• A deleted value always reads as absent. Time travel does not restore it. A restarted SubFlow starts from the original initial value, so it reads as absent when an earlier run deleted that value.
• Removing the option from a definition leaves the stored value in place until Dex removes the Flow. Delete the value with the hard-delete definition first.

A value remains stored until it is deleted, or until Dex removes the Flow after its retention period. Object storage backups and bucket versioning outside Dex can keep copies. See Object storage.

Erase after the run closes

Since server-zd 0.1.5, a direct call to the FlowService SetAttributes API can erase a hard-delete Attribute after its Flow run closes and before the run is purged. The SDK Clients do not expose this API, so send it with dexcli api call SetAttributes or another FlowService client. Set every Attribute in the request to null. Each Attribute must still have a stored hard-delete value. The erase leaves the run closed and permanently deletes its stored objects. Other writes to a closed run return NOT_FOUND.

Locks

Dex Server lets only one Step execution or RPC invocation hold an Attribute lock at a time. A Step declares locks separately for WaitFor and Execute in its Step options. An RPC declares its locks when it is registered. The Attribute page shows an Execute lock.

A Step attempt takes its locks when Dex hands it to a Worker. A first attempt that no Worker took yet holds no locks, but later Steps that need the same lock wait behind it. A returned attempt, for example after a Worker declined its Step type, and a retry attempt keep their locks while they wait in the queue. When no Worker can run them, they hold the locks until the Flow times out or stops. Transactional RPCs on those keys get the lock conflict until the attempt runs and finishes.

Lock WaitFor:

def get_step_options(self) -> StepOptions:
    return StepOptions(
        wait_for_lock_attributes=(
            self.flow.status.lock(),
            self.flow.progress.lock("payment"),
        ),
    )

Example: examples/python/dex_examples/primitives/attribute/attribute_flow.py

Lock an RPC:

@rpc(lock_attributes=(status.lock(), progress.lock("payment")))
def update_status(self, context: Context, input: str) -> RPCResult[str]:
    self.status.set(context, input)
    self.progress.set(context, "payment", input)
    return RPCResult(input)

Example: examples/python/dex_examples/primitives/attribute/attribute_flow.py

Coordinate writes across the whole AttributeMap

An AttributeMap lock protects one exact instance. When an invariant spans several instances or the whole collection, register a separate singleton bool Attribute as its coordination key. For example, register currentMessages as an AttributeMap and currentMessagesLock as a non-map, non-indexed bool Attribute in the same PersistenceSchema.

Every Step, timeout handler, and RPC that writes, deletes, archives, or otherwise changes the protected map must request the same singleton Attribute lock in its method-specific lock options. RPCs declare this fixed lock at registration time. A writer that requests only a map-instance lock does not conflict with the singleton lock. All participating writers serialize within that Flow execution; other Flow executions have their own locks.

The bool value is not the lock state. Do not acquire or release the lock by reading or toggling the value. Dex acquires and releases it around the invocation; the coordination Attribute does not need a persisted value. Locking also does not load map entries. Declare exact instance loads, or a whole-map load when the operation must enumerate instances. RPC lock acquisition can return a lock-conflict error rather than queueing the caller.

Keep per-instance locks for independent partition updates. Use the shared singleton lock when all affected writers must preserve a collection-wide invariant.

RPC lock conflicts

RPC lock acquisition does not queue callers. When another Step execution or RPC holds a requested lock, the call fails with the error "RPC lock conflict: attribute key <key> is locked by a Step". The SDK exposes it as RPCLockConflictError in Go, RpcLockConflictError in Python and TypeScript, RpcLockConflictException in Java, and SdkError::RpcLockConflict in Rust.

Retry specifically after that error, with application-appropriate backoff. Make the RPC idempotent so that a retry is safe. Successful calls serialize through the same locks, so concurrent requests do not lose or duplicate updates.

RPCs

Language-specific RPC rules are in the SDK references: Java Flow classes and Kotlin, and TypeScript codecs.

Cancel running Steps

An RPC result can cancel active Steps of its Flow. Select the Steps by Step type: with_canceling_steps in Python, CancelSteps in Go, withCancelingSteps in Java, cancelingSteps in TypeScript, and cancel_step in Rust. The selection covers the whole Flow. An RPC has no parent Step execution, so it cannot cancel sibling Steps. A Step can cancel other Steps the same way from its StepDecision.

In the example, GenerateReport writes a report section by section. The stop_report RPC cancels it and starts FinishReportStop.

@rpc(timeout=timedelta(seconds=30))
def stop_report(self, context: Context, input: None) -> RPCResult[bool]:
    if REPORT_STATUS.get(context) in (REPORT_READY, REPORT_STOPPED):
        return RPCResult(False)
    return RPCResult(
        True,
        next_steps=(StepMovement.of(FinishReportStop, None),),
    ).with_canceling_steps(GenerateReport)

Example: examples/python/dex_examples/primitives/rpc_canceling_steps

Dex applies the cancellation when it commits the RPC. It commits the RPC's Attribute writes and Channel publications first, then cancels, then starts the RPC's next Steps. All of it commits together, or none of it does. The RPC's own next Steps are not canceled, even when they have a selected Step type. A selected Step that has already finished is skipped, and so is a type with no active Step.

Cancellation ends every active Step of a selected type: one that waits, one that waits for a retry, and one whose attempt runs on a Worker. For a running attempt, Dex sends a cancellation to that Worker. A canceled Step is not retried, and its failure route does not run. Dex discards anything the canceled attempt returns later, including its StepDecision and Attribute writes.

Resolve the race from committed state

The RPC handler reads a snapshot of the Flow when it starts. Dex commits its result later. The Steps to cancel are chosen at commit time, not from the snapshot. Between the two, a selected Step can finish and commit its own result. Then the cancellation finds nothing to cancel. A new Step of a selected type can also start, and the cancellation ends it too. Dex does not reject the RPC in either case. So the handler cannot assume that the Step it saw is still running, or that its work did not commit.

Do not decide the outcome in the RPC handler. Start a follow-up Step from the RPC, and let it read the committed state. In the example, FinishReportStop checks whether the report committed before the cancellation. If it did, the Stop came too late, and the Flow delivers the report. Otherwise, it records that the report stopped.

def execute(self, context: Context, input: None) -> StepDecision:
    # The report can commit after stop_report read its snapshot.
    if REPORT_STATUS.get(context) == REPORT_READY:
        return dead_end()
    REPORT_STATUS.set(context, REPORT_STOPPED)
    return graceful_complete(REPORT_STOPPED)

Example: examples/python/dex_examples/primitives/rpc_canceling_steps

Attribute locks do not close this gap on their own. They serialize the RPC only with Step methods that declare the same locks.

Make the Step stop

Dex cannot stop code that ignores cancellation. The Step code must check for it and return:

• Python: with AsyncWorker, Dex cancels the handler's task, so the next await raises CancelledError. With Worker, the handler runs on a thread and must check is_cancellation_requested.
• Go: Dex cancels the handler's Context. Pass it to calls that accept a context, or select on Done.
• Java: Dex interrupts the handler thread. Blocking calls throw InterruptedException, and isCancellationRequested returns true.
• TypeScript: Dex aborts context.cancellationSignal. Pass it to calls that accept an AbortSignal.
• Rust: check is_cancelled, or block in wait_for_cancellation.

for section in range(1, SECTION_COUNT + 1):
    # A Stop cancels this task, so the await raises CancelledError.
    await asyncio.sleep(SECTION_WORK.total_seconds())
    REPORT_PROGRESS.write(context, section)

Example: examples/python/dex_examples/primitives/rpc_canceling_steps

Dex waits about one second for the canceled attempt to end. If the handler is still running, Dex closes the Worker stream that carried it. The handler keeps running inside the Worker process until it returns, and its external calls still take effect. A canceled attempt may also have done part of its work before it stopped. Keep its external effects idempotent. See Failure and redelivery.

Read-only RPCs and terminal Flows

To read stored Attribute values from outside the Flow, an application does not need a read-only RPC. The Client's GetAttributes reads committed values directly, without a Worker. See Read Attributes from outside a Flow. Use a read-only RPC when the read needs Flow-side logic, such as validation, authorization, Channel messages, or a derived view.

An RPC without Attribute locks and without transactional execution starts with a read of the Flow. If its handler returns only an output and no durable effects, Dex returns that output without changing the Flow. That read can reach a retained terminal execution. A successful read-only RPC therefore does not prove that the Flow is active.

Attribute locks and explicit transactional execution use an active execution. A non-transactional handler that returns Attribute writes, Channel operations, events, or Step movements also needs an active execution because Dex commits those effects to the Flow after the handler returns. A terminal Flow rejects that commit even though the Worker handler may already have run. A Server can also be configured to run every RPC transactionally, which requires an active execution.

Use a Flow lifecycle API when the response needs current execution status. An application snapshot RPC reads the durable application state captured by the Flow; it does not infer terminal status from RPC success. FlowNotActiveOrNotFoundError means the target was not found, or the selected operation cannot use its closed execution. Java uses FlowNotActiveOrNotFoundException; Rust uses SdkError::FlowNotActiveOrNotFound. The error alone does not distinguish missing from closed Flows.

For a confirmed read-only RPC with no locks, transactionality, durable effects, or Server-forced transactional execution, a retained terminal execution remains readable. At that read boundary, handle this error directly as no readable target found. Preserve the application's missing or retention-unavailable outcome. Do not add a lifecycle lookup, WaitForFlow with a short timeout, a retry, or historical Step-output decoding solely to distinguish missing from closed. Keep other service and Worker failures distinct.

For mutations and active-only RPCs, the same error may mean the Flow is closed or missing. Apply the operation's business policy; the error does not prove that the requested action succeeded.

Callers do not need to search for a Run ID before a normal read-only RPC. When the SDK exposes an optional Run ID, omit it to let the Server resolve and pin the current execution for that invocation. Supply one only when the application must address an exact execution across Flow ID reuse.

Selective state loading

Every RPC receives regular Attribute values and Channel size metadata. Reading AttributeMap entries or pending Channel messages requires an explicit load. This keeps large collections out of RPC requests that do not use them.

Load a whole AttributeMap or ChannelMap when the handler needs all current instances. Load exact map instances when it needs only known keys. Load a Channel when the handler needs its pending message IDs or Values. Channel size and ChannelMap keys and sizes remain available without loading pending messages.

Every SDK declares loads with the RPC. Go places them in RPCOptions passed to DefineRPC from GetRPCs. Dex Server validates each requested definition against the Flow persistence schema and returns the loaded scope with the RPC snapshot. The SDK checks that scope when the handler reads a selectively loaded collection.

For an AttributeMap entry or instance enumeration outside that scope, the SDK raises AttributeMapNotLoadedError. For pending messages outside that scope, it raises ChannelMessagesNotLoadedError. Java uses AttributeMapNotLoadedException and ChannelMessagesNotLoadedException. For Go, dexcli visualize reports an RPC that reads unselected state. See State loads.

An explicitly loaded collection can be empty. That is not an error. An empty pending-message snapshot returns an empty list. An absent AttributeMap entry uses the SDK's normal absent-value behavior. Loading creates one input snapshot. It does not consume messages, make the RPC transactional, or isolate the handler from concurrent work.

Runtime-selected map instances

Use the RPC definition options for fixed behavior shared by every invocation: timeout, ordinary Attribute locks, transactionality, whole-map loads, singleton Channel loads, and fixed exact-instance loads. Go names this type RPCOptions.

Use RPCInvokeOptions when the application adapter chooses an AttributeMap or ChannelMap instance from request data. Rust uses RpcInvokeOptions. Invocation options can add exact AttributeMap loads, exact ChannelMap loads, and AttributeMap instance locks. Dex unions, sorts, and deduplicates them with the registered definition. They cannot replace or clear registered settings.

Loading and locking are independent. A read-only lookup needs only the exact load. A read-modify-write RPC must add both the load and the lock for the same AttributeMap instance. The SDK validates the Flow ownership, definition kind, and non-empty slash-free instance before making the network call.

options=RPCInvokeOptions(
    lock_attribute_map_instances=(
        flow.customer_profiles_by_email_partition.lock(partition_name),
    ),
    load_attribute_map_instances=(
        flow.customer_profiles_by_email_partition.load(partition_name),
    ),
)

Example: examples/python/dex_examples/patterns/hash-partitioned-attribute-map/controller.py

See Hash-Partitioned AttributeMap for the complete lookup and upsert example, and Sequentially Chunked AttributeMap for dynamic pagination loads.

Transactional RPCs

A transactional RPC performs its reads and writes as one atomic operation. An RPC that holds an Attribute lock is already transactional implicitly. You can also make an RPC explicitly transactional by setting is_transactional=true.

Transactional execution matters most for Channel message deletion. Dex validates that every deleted message is still pending before it commits any write. If a message is gone, the RPC fails with the Channel-message-not-found error (NOT_FOUND) and commits nothing. A non-transactional RPC treats a missing deletion as a no-op and still commits its other successful effects.

In the example, the caller passes only a pending message ID. The RPC loads the source Channel, reads the original Value from its snapshot, then stages the deletion and destination publication. A concurrent consumer makes deletion validation fail, so Dex commits neither effect:

@rpc(is_transactional=True, load_channels=(queued_messages,))
def move_queued_message_to_prioritized_messages(
    self, context: Context, queued_message: QueuedMessageReference
) -> None:
    message_to_prioritize = self.queued_messages.find_pending_message(
        context, queued_message.message_id
    )
    self.queued_messages.delete(context, queued_message.message_id)
    if message_to_prioritize is not None:
        self.prioritized_messages.publish(context, message_to_prioritize.value)

Example: examples/python/dex_examples/primitives/channel/channel_flow.py

A transactional commit does not isolate the handler while it runs. If a decision depends on the whole loaded Channel, ChannelMap, or AttributeMap remaining unchanged, every cooperating Step and RPC writer must use the same Attribute lock. The lock provides that isolation and implicitly makes the RPC transactional; it does not load collection contents.

RPC timeout

An RPC timeout limits one Worker handler invocation.

Every SDK declares the timeout with the RPC. Go places it in the static RPCOptions passed to DefineRPC. A Client invocation cannot override it, although the caller's context can stop waiting sooner. When the application does not set a timeout, Dex Server uses its default of 60 seconds, set by api.defaultRpcTimeout. An explicit timeout has no maximum.

@rpc(timeout=timedelta(seconds=30))

Example: examples/python/dex_examples/primitives/rpc

Channels

Wait conditions

AtMost does not wait on its own for messages to accumulate. When the surrounding Wait completes, it consumes up to the limit from the queue at that time. To wait for at least one message and consume up to a limit, use AtLeastAtMost with a lower bound of 1.

Pending messages

Dex assigns a UUIDv7 message ID when it accepts each publication. A Step or RPC can list every pending message in FIFO order and delete one by ID. Listing does not consume messages.

Deleting a message that is no longer pending depends on the caller:

Caller	Missing message
Transactional RPC	The RPC fails with the Channel-message-not-found error and commits nothing. See Transactional RPCs.
Non-transactional RPC	The deletion is a no-op. The other effects commit.
Step method or Flow timeout handler	The deletion is a no-op. The other effects commit.
Direct DeleteChannelMessage (dexcli, Dex Web)	The call fails with NOT_FOUND and sub-status CHANNEL_MESSAGE_NOT_FOUND.

Pending messages are queue state, not Flow history. After a Step consumes a message, it disappears from the Channel even though the publication and consumption remain observable in recent Flow history.

Every Step method, timeout handler, and RPC always receives Channel size metadata. It must explicitly load pending messages before reading their IDs or Values. Deleting a message whose ID is already known does not require loading it. StepOptions selects state separately for WaitFor and Execute; FlowTimeoutHandlerOptions selects state for the timeout handler. Loading a Channel creates one invocation snapshot and does not consume messages. A loaded empty Channel returns an empty list; reading pending messages without loading the Channel is a usage error.

Treat every pending-message snapshot read inside a Step or RPC as potentially stale. Other Steps and RPCs can consume, delete, or publish messages while the handler is running. Transactional execution validates selected deletions and commits the handler's writes atomically, but it does not lock the whole Channel snapshot. Read and write pending messages directly only when the operation explicitly tolerates that concurrency. If a decision requires the queue to remain unchanged, every cooperating Step and RPC writer must use the same Attribute lock.

A Step or timeout handler can stage deletion with its other side effects. Dex applies successful handler effects in this order: Attribute writes, best-effort Channel deletions, then Channel publications, followed by the Step decision. StepDecision describes only control flow; it does not contain deletions.

An RPC can combine deletion with other durable writes. The move example in Transactional RPCs loads QueuedMessages, finds the original Value by message ID, then stages its deletion and republishes that Value. The caller sends only the ID.

ChannelMap

RPCs can always read the current ChannelMap keys and sizes. Load the entire ChannelMap when the handler needs every instance's pending messages. Load exact instances when it needs only known keys. Instance keys cannot contain / because Dex uses that character between the map name and instance key.

Streams

Server settings for Streams, with their defaults, are in Streams.

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 Polling pattern uses these writes to report progress.
• 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. After a Dex Server node crashes, an ASYNC Step can also run the same attempt number again, and its batches can repeat in the same source.

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.

List 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 page size must be between 1 and streamStore.maxReadMessages, which defaults to 1000. The server rejects a page size outside that range with INVALID_ARGUMENT; it does not reduce a larger value.

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.

Capacity and trimming

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, background trimming removes the oldest messages across all instances until usage reaches the trim target threshold. A server runs several trim workers, but each Stream is trimmed by one job at a time. If trimming is too slow and usage reaches the capacity limit (streamCapacityBytes), the write request is rejected with RESOURCE_EXHAUSTED. The rejection also asks trimming to free room for that message, so a Client can retry the write. A message larger than the Stream's capacity is rejected the same way, but it can never fit, so do not retry it.

The server also limits the size of each Stream message, the capacity of one Stream, and the memory of all Streams on a server. A Client write past the server-wide total fails with RESOURCE_EXHAUSTED, and background trimming then removes the oldest messages of the largest Streams until the message fits. Retry the write after a short delay. See Streams for each limit and its default.

Best-effort storage

The Stream Store keeps messages in the memory of the active server process. streamStore.backend defaults to memory for both dexcli dev and a production server. Memory contents disappear when the server process restarts or a standby takes over. Trimming or capacity pressure can also remove or reject messages. An operator can set the backend to disabled. Client Stream writes and reads then fail with FAILED_PRECONDITION, and Step writes are dropped without failing the Step.

SubFlows

SubFlowOptions

SubFlowOptions are the equivalent of StartFlowOptions on the StartFlow API. Most options are the same, except for SubFlowReusePolicy.

SubFlowReusePolicy

Dex Server generates a SubFlow ID from the parent Flow ID, Step execution ID, and condition index. When a parent reuses a Flow ID with IDReusePolicy, or time travels, it can enter the same Step execution and try to start the same SubFlow ID again. SubFlowReusePolicy controls what happens in that case.

The default RESTART_IF_PREVIOUS_EXITS_ABNORMALLY is usually what a parent wants: continue waiting for a running child, use a successful child result, and restart a child that ended abnormally.

Existing child Flow	ATTACH	RESTART_IF_PREVIOUS_EXITS_ABNORMALLY	ALWAYS_RESTART
No existing Flow	Start	Start	Start
Running	Attach	Attach	Terminate and start
Completed	Return its result	Return its result	Start
Failed, canceled, timed out, or terminated	Return its result	Start	Start

Dex Server first compares the Request ID it generated for this child start. A matching Request ID always attaches to a running child or returns its terminal result, regardless of ReusePolicy. This makes a retry safe when Dex Server accepted the original start but the parent did not receive the response. For a different Request ID, Dex Server applies the selected ReusePolicy from the table.