mstar.streaming.chunk_policy

Classes

ChunkPolicy()	Determines when a StreamBuffer has enough items for the consumer node.
FixedChunkPolicy(chunk_size[, ...])	Release non-overlapping chunks of fixed size.
LeftContextChunkPolicy(chunk, left_context)	Chunk policy for streaming vocoders with left-context overlap.
RampChunkPolicy(first_chunk, chunk_size[, ...])	A smaller first chunk, then chunks that grow geometrically up to a cap.
ScheduledLeftContextChunkPolicy(schedule, ...)	Left-context chunking whose chunk sizes follow a ramp.
SlidingWindowChunkPolicy(window, stride)	Fixed-size sliding window that advances by a stride.

class mstar.streaming.chunk_policy.ChunkPolicy

Bases: ABC

Determines when a StreamBuffer has enough items for the consumer node.

continue_after_producer_done()

Whether the buffer should keep producing (empty) chunks after the producer signals done and all buffered items have been consumed.

Default False: partition-done is propagated to the conductor after the last item is flushed.

Set to True for connections where the consumer must keep running after the producer finishes (e.g., Thinker→Talker: the Talker continues generating codec tokens after the Thinker hits text EOS). In this case the buffer produces empty chunks (_collate([]) → {"data": None}), and the consumer’s partition-done is determined by its own model logic, not by the StreamBuffer.

Return type:

bool

continues_in(consumer_walk)

Whether empty chunks keep coming, after the producer is done, while the consumer is in consumer_walk.

Parameters:

consumer_walk (str | None)

Return type:

bool

abstractmethod is_ready(buffer_len)

Return True if the buffer has enough items for a chunk.

Parameters:

buffer_len (int)

Return type:

bool

abstractmethod next_chunk_size(buffer_len)

Return the number of items to consume for the next chunk.

Only called when is_ready() returns True. For sliding-window policies this is the stride, not the window.

Parameters:

buffer_len (int)

Return type:

int

prime(context_items)

The stream starts with context_items items that count as already delivered: they reach the consumer only as context of the first chunk.

Called before the first chunk. Only policies with a notion of context support it.

Parameters:

context_items (int)

Return type:

None

register_chunk(chunk_size)

Parameters:

chunk_size (int)

abstractmethod window_size()

Return the full window of items to include in the chunk.

For non-overlapping policies, equals next_chunk_size. For sliding-window policies, this is larger than the stride — the buffer retains older items so the chunk contains the full window.

Return type:

int

class mstar.streaming.chunk_policy.FixedChunkPolicy(chunk_size, continue_after_done=False)

Bases: ChunkPolicy

Release non-overlapping chunks of fixed size.

Each pop_chunk returns exactly chunk_size items and advances by chunk_size. No overlap, no sliding window.

Parameters:

• chunk_size (int) – number of items per chunk.

• continue_after_done (bool | frozenset[str]) – if True, keep producing empty chunks after the producer finishes and all buffered items are consumed. A set of consumer walks does so only while the consumer is in one of them; in its other walks the stream just waits.

continue_after_producer_done()

Whether the buffer should keep producing (empty) chunks after the producer signals done and all buffered items have been consumed.

Default False: partition-done is propagated to the conductor after the last item is flushed.

Set to True for connections where the consumer must keep running after the producer finishes (e.g., Thinker→Talker: the Talker continues generating codec tokens after the Thinker hits text EOS). In this case the buffer produces empty chunks (_collate([]) → {"data": None}), and the consumer’s partition-done is determined by its own model logic, not by the StreamBuffer.

Return type:

bool

continues_in(consumer_walk)

Whether empty chunks keep coming, after the producer is done, while the consumer is in consumer_walk.

Parameters:

consumer_walk (str | None)

Return type:

bool

is_ready(buffer_len)

Return True if the buffer has enough items for a chunk.

Return type:

bool

next_chunk_size(buffer_len)

Return the number of items to consume for the next chunk.

Only called when is_ready() returns True. For sliding-window policies this is the stride, not the window.

Parameters:

buffer_len (int)

Return type:

int

window_size()

Return the full window of items to include in the chunk.

For non-overlapping policies, equals next_chunk_size. For sliding-window policies, this is larger than the stride — the buffer retains older items so the chunk contains the full window.

Return type:

int

class mstar.streaming.chunk_policy.LeftContextChunkPolicy(chunk, left_context)

Bases: ChunkPolicy

Chunk policy for streaming vocoders with left-context overlap.

Matches HuggingFace’s Qwen3OmniMoeCode2Wav.chunked_decode pattern:

Iter 0: codes[0 : chunk] → emit all (no context) Iter 1: codes[chunk-ctx : 2*chunk] → trim first ctx, emit rest Iter 2: codes[2*chunk-ctx : 3*chunk] → trim first ctx, emit rest

The first pop returns chunk items (no context). Subsequent pops return chunk + left_context items, where the leading left_context items OVERLAP with the tail of the previous chunk. This overlap allows the causal ConvNet vocoder to “warm up” its internal state on frames it has already processed, ensuring a smooth transition at chunk boundaries.

The key invariant: the first pop advances by chunk - left_context (not chunk), so the last left_context items of the first chunk remain in the buffer as overlap for the second pop. All subsequent pops advance by chunk.

Parameters:

• chunk (int)

• left_context (int)

is_ready(buffer_len)

Return True if the buffer has enough items for a chunk.

Parameters:

buffer_len (int)

Return type:

bool

next_chunk_size(buffer_len)

Return the number of items to consume for the next chunk.

Only called when is_ready() returns True. For sliding-window policies this is the stride, not the window.

Parameters:

buffer_len (int)

Return type:

int

window_size()

Return the full window of items to include in the chunk.

For non-overlapping policies, equals next_chunk_size. For sliding-window policies, this is larger than the stride — the buffer retains older items so the chunk contains the full window.

Return type:

int

class mstar.streaming.chunk_policy.RampChunkPolicy(first_chunk, chunk_size, continue_after_done=False, growth=1.0, max_chunk=None)

Bases: ChunkPolicy

A smaller first chunk, then chunks that grow geometrically up to a cap.

A streaming consumer that re-runs over the whole accumulated stream on every chunk (a flow-matching mel decoder conditioned on all tokens so far) wants its first chunk as early as the model allows, for time-to-first-audio, and later chunks sized for throughput. first_chunk items release the first chunk; the next is chunk_size items and every chunk after that is growth times the previous one, capped at max_chunk (growth=1 keeps them fixed). Growing chunks track the listener’s playback buffer: each chunk buys the time to produce a bigger one, so fewer, larger solves carry the same stream. No overlap: the consumer keeps its own history.

continue_after_done behaves as in FixedChunkPolicy.

Parameters:

• first_chunk (int)

• chunk_size (int)

• continue_after_done (bool)

• growth (float)

• max_chunk (int | None)

continue_after_producer_done()

Whether the buffer should keep producing (empty) chunks after the producer signals done and all buffered items have been consumed.

Default False: partition-done is propagated to the conductor after the last item is flushed.

Set to True for connections where the consumer must keep running after the producer finishes (e.g., Thinker→Talker: the Talker continues generating codec tokens after the Thinker hits text EOS). In this case the buffer produces empty chunks (_collate([]) → {"data": None}), and the consumer’s partition-done is determined by its own model logic, not by the StreamBuffer.

Return type:

bool

is_ready(buffer_len)

Return True if the buffer has enough items for a chunk.

Parameters:

buffer_len (int)

Return type:

bool

next_chunk_size(buffer_len)

Return the number of items to consume for the next chunk.

Only called when is_ready() returns True. For sliding-window policies this is the stride, not the window.

Parameters:

buffer_len (int)

Return type:

int

register_chunk(chunk_size)

Parameters:

chunk_size (int)

window_size()

Return the full window of items to include in the chunk.

For non-overlapping policies, equals next_chunk_size. For sliding-window policies, this is larger than the stride — the buffer retains older items so the chunk contains the full window.

Return type:

int

class mstar.streaming.chunk_policy.ScheduledLeftContextChunkPolicy(schedule, chunk, left_context)

Bases: ChunkPolicy

Left-context chunking whose chunk sizes follow a ramp.

Streaming vocoders want the first audio out as early as possible and larger chunks once the stream is running. Chunk k delivers schedule[k] new items (chunk once the schedule is exhausted) with up to left_context already-delivered items in front of them, so a causal decoder can warm up on frames it has processed before. Unlike LeftContextChunkPolicy the first chunk may be smaller than the context: the context is whatever has been delivered so far, capped.

Example (Qwen3-TTS, 12 Hz frames): schedule=(4, 8, 16), chunk=25, left_context=25 pops windows of 4, 4+8, 12+16, 25+25, 25+25, … items and the first audio leaves after four frames instead of 300.

The consumer learns how many leading items of a window are context from StreamChunk.context_items (the worker passes it along as input_metadata.stream_chunks[edge].context_items), so it can trim the duplicated output without re-deriving this schedule.

Parameters:

• schedule (Sequence[int])

• chunk (int)

• left_context (int)

is_ready(buffer_len)

Return True if the buffer has enough items for a chunk.

Parameters:

buffer_len (int)

Return type:

bool

next_chunk_size(buffer_len)

Return the number of items to consume for the next chunk.

Only called when is_ready() returns True. For sliding-window policies this is the stride, not the window.

Parameters:

buffer_len (int)

Return type:

int

prime(context_items)

The stream starts with context_items items that count as already delivered: they reach the consumer only as context of the first chunk.

Called before the first chunk. Only policies with a notion of context support it.

Parameters:

context_items (int)

Return type:

None

register_chunk(chunk_size)

Parameters:

chunk_size (int)

window_size()

Return the full window of items to include in the chunk.

For non-overlapping policies, equals next_chunk_size. For sliding-window policies, this is larger than the stride — the buffer retains older items so the chunk contains the full window.

Return type:

int

class mstar.streaming.chunk_policy.SlidingWindowChunkPolicy(window, stride)

Bases: ChunkPolicy

Fixed-size sliding window that advances by a stride.

Each pop_chunk returns window items and advances the consumed pointer by stride. Old items before the window are discarded.

Example (Orpheus SNAC): window=28 tokens (4 frames), stride=7 (1 frame).

Parameters:

• window (int)

• stride (int)

is_ready(buffer_len)

Return True if the buffer has enough items for a chunk.

Parameters:

buffer_len (int)

Return type:

bool

next_chunk_size(buffer_len)

Return the number of items to consume for the next chunk.

Only called when is_ready() returns True. For sliding-window policies this is the stride, not the window.

Parameters:

buffer_len (int)

Return type:

int

window_size()

Return the full window of items to include in the chunk.

For non-overlapping policies, equals next_chunk_size. For sliding-window policies, this is larger than the stride — the buffer retains older items so the chunk contains the full window.

Return type:

int