Streaming

Multiplexed streams, chunking, sequence numbering, and reordering.

Multiplexed streams

A single request (identified by its request ID in key 2) can carry multiple named streams. Each stream has a unique stream_id (key 11) and a media_urn (key 12) that identifies the data type flowing through it.

This enables multi-argument capability invocations. For example, a cap that accepts both a PDF document and a prompt string receives them as two separate streams within the same request: one with media_urn = "media:document;ext=pdf" and one with media_urn = "media:enc=utf-8". On the response side, a handler can emit multiple output streams too.

Source: capdag-rs/src/bifaci/frame.rs (StreamStart, StreamEnd); cartridge_runtime.rs (InputStream, OutputStream).

Stream lifecycle

A stream follows this sequence:

sequenceDiagram
    participant S as Sender
    participant R as Receiver

    S->>R: STREAM_START (stream_id, media_urn)
    S->>R: CHUNK (chunk_index=0, checksum)
    S->>R: CHUNK (chunk_index=1, checksum)
    Note over S,R: ... more chunks ...
    S->>R: CHUNK (chunk_index=N, checksum)
    S->>R: STREAM_END (chunk_count=N+1)

• STREAM_START announces the stream. The media_urn tells the receiver what kind of data to expect.
• CHUNK frames carry payload data. Each has a chunk_index starting at 0 and a checksum for integrity.
• STREAM_END marks the stream as complete. Its chunk_count field holds the total number of CHUNKs sent, allowing the receiver to verify it received them all. Unbounded streams omit it (below).

After STREAM_END, any CHUNK for the same stream_id is a protocol error.

Request completion

After all streams within a request have ended (each with its own STREAM_END), the sender sends END to signal that no more streams will be opened and the request is complete:

sequenceDiagram
    participant S as Sender
    participant R as Receiver

    rect rgb(240, 248, 255)
        Note right of S: Stream A
        S->>R: STREAM_START (stream A)
        S->>R: CHUNK(s)
        S->>R: STREAM_END (stream A)
    end

    rect rgb(245, 255, 245)
        Note right of S: Stream B
        S->>R: STREAM_START (stream B)
        S->>R: CHUNK(s)
        S->>R: STREAM_END (stream B)
    end

    S->>R: END

No flow frame may follow an END for the same request ID (L4). Runtimes enforce this at their write stage: a post-terminal flow frame is a BENIGN straggler: suppressed and counted per frame type in the straggler counters (never a drop, since nothing went wrong) and never written. Terminal metadata (the final progress and message) rides in the END frame itself (L5), so nothing important can race the terminal. Intermediaries forward the terminal before releasing routing state; frames arriving for released state are counted benign stragglers when the ledger vouches for the termination, and no_route drops otherwise, never protocol errors (L6).

ERR is an alternative terminal frame. It signals that the request failed. Like END, no data frames may follow it. ERR and END are mutually exclusive: a request ends with exactly one.

LOG frames (progress updates, log messages) can appear anywhere in the sequence, interleaved with CHUNK and between streams. They do not affect the data flow.

Unbounded streams

A STREAM_START may carry unbounded = true (key 20): the stream makes no length promise (L16). A live capture, a generated video, an open-ended transcription: anything whose size is unknown when the stream opens.

• The producer opens it with OutputStream::start_unbounded(); its STREAM_END omits chunk_count.
• Receivers MUST consume unbounded streams incrementally (InputStream::recv, TerminalOutput::next_item). The buffering collectors (collect_bytes, collect_items, collect_value) refuse unbounded streams with a hard error, because buffering a stream with no length promise is unbounded memory.
• “Unbounded” means “no promised length”, not “never ends”: the stream still terminates via STREAM_END, the request terminal, or Cancel. Cancel is effective mid-stream (L17): producer and consumer both observe termination promptly (blocked reads unblock, credit waiters release).
• Flow control is what makes this safe: the producer can never run more than the granted window ahead of the consumer. See Flow control.
• An unbounded emission is declared before it happens: a cap may only open an unbounded output stream if its definition’s output carries streaming: true, and it may only be wired live to an unbounded producer if its main input carries streaming: true (15.2 §Streaming contracts, 10 RULE14). The executor audits every received STREAM_START against that declaration: an unbounded flag from a non-streaming output, or an is_sequence that differs from the declared cardinality, fails the run at receipt (StreamContractViolation). The declaration is also what decides whether a hop pipelines or materialises, so a buffering consumer never meets an unbounded stream by surprise.

Live feeds

A live feed (microphone, webcam, any real-time capture) is an unbounded sequence stream produced by the runtime’s transport-resolution layer from a live-feed reference arg (see Input and output §Reference media). It is not a new stream kind; it is the unbounded + sequence combination this section already defines, with a metadata contract:

• Stream-level meta (STREAM_START): the capture’s negotiated format actuals, the parameters the device actually opened with (sample rate, channels, resolution, pixel format), which may be more specific than the content URN’s tags.
• Per-item meta (the emit_list_item meta slot): every item carries
  • seq: the capture-order item index, monotonic from 0, counting dropped items too (a gap in delivered seq is real),
  • pts_us: presentation timestamp, microseconds from capture start,
  • capture_ts_us: wall-clock capture time (Unix µs),
  • gap: present only on the first item after an overrun: { "dropped": N, "duration_us": D }, the count and time span of items discarded at the capture edge since the previously delivered item. Downstream MUST treat a gap as a genuine discontinuity (e.g. a transcriber resets its phrase state), never silently concatenate across it.

Ordering is layered: (1) per-flow wire order is transport-guaranteed (SeqAssigner + ReorderBuffer); (2) item boundaries within a stream are chunk-run delimited, so item order within one stream is end-to-end preserved; (3) the per-item seq/pts_us exist for the one place order genuinely scrambles (parallel ForEach re-join) and for gap-aware time reconstruction. The wire never reorders; parallelism does.

Sequence items are never write-coalesced (the coalescing exemption), so a live item is on the wire the moment it is emitted, and feed latency is one item, not one coalescing window.

Bounded inputs are the degenerate case of this model: a file is a live feed whose stream ends immediately after its content. The machinery is identical; only the moment END arrives differs.

Chunking

Automatic chunking

When the payload to send exceeds max_chunk (negotiated during handshake, default 256 KB), the writer splits it automatically. The FrameWriter::write_chunked() method handles this:

1. The first chunk (chunk_index = 0) carries content_type, len (total byte count), and offset = 0.
2. Subsequent chunks carry offset (the byte position within the total payload).
3. The final chunk has eof = true.

An empty payload is sent as a single chunk with eof = true and len = 0.

From the handler’s perspective, this is transparent. The OutputStream::write() method calls write_chunked internally when the data exceeds the chunk size limit. The receiver’s InputStream reassembles chunks in order.

Source: capdag-rs/src/bifaci/io.rs (FrameWriter::write_chunked); cartridge_runtime.rs (OutputStream::write).

Sequence-mode item framing

The two stream modes frame their CHUNK payloads differently, and the receiver’s demux must decode accordingly:

• Scalar mode (is_sequence = false, OutputStream::write): every CHUNK payload is a self-contained CBOR value: raw bytes are split at max_chunk boundaries and each piece is re-wrapped as its own Value::Bytes. Per-frame decoding is always valid. Chunk boundaries carry no meaning of their own (receivers concatenate the inner bytes), which is what makes write coalescing legal: runtimes fold small scalar byte writes into one CHUNK, flushed when the batch reaches 4096 bytes, when the oldest buffered byte is 20 ms old at the next write, at any non-Bytes emission on the same stream (an ordering barrier), and unconditionally by close() BEFORE the STREAM_END whose chunk_count therefore promises the coalesced count. A per-token LLM emitter thus produces at most ~one frame per age window during steady generation instead of one frame per token, dividing frame and credit traffic by the batch factor without any receiver-visible change. Sequence streams are NEVER coalesced.
• Sequence mode (is_sequence = true, OutputStream::emit_list_item): each item is CBOR-encoded once, and the encoded bytes are split across CHUNK frames at max_chunk boundaries. A frame payload is a raw RFC 8742 fragment, not a self-contained CBOR value. Per-item metadata rides the item’s first fragment frame only. Frames never straddle an item boundary: every item starts on a fresh frame.

The receiver therefore reassembles sequence streams at item granularity: fragments are buffered per stream and decoded incrementally (CBOR’s definite-length encoding is prefix-free, so an incomplete item is unambiguously “need more bytes”, never a mis-decode). One decoded value is delivered to the handler per item, carrying the first fragment’s metadata. A STREAM_END arriving with buffered bytes that do not complete an item is a hard decode error (truncated item), never a silent drop.

Credit interaction (L10/L12): the credit window is frame-granular, but the handler consumes and grants per delivered item. The demux credits every continuation fragment (all frames after an item’s first) back to the sender immediately on arrival, so an item spanning more frames than the credit window can always finish arriving; the item’s first frame remains covered by the handler’s consumption grant, preserving item-level backpressure.

Source: cartridge_runtime.rs (emit_list_item, SeqReassembly, try_decode_sequence_item, demux_multi_stream, demux_single_stream).

Chunk fields

Every CHUNK frame carries:

The chunk_index and checksum fields are validated at decode time, and a CHUNK frame missing either is rejected with a protocol error.

Sequence numbering

Frames within a request need to arrive in order for the protocol to work correctly. At relay boundaries, frames from different masters may interleave or arrive out of order. Sequence numbers enable detection and correction of this.

FlowKey

A flow is identified by the pair (request_id, optional routing_id), called a FlowKey. Each unique FlowKey gets its own independent sequence counter.

pub struct FlowKey {
    pub rid: MessageId,       // request ID (key 2)
    pub xid: Option<MessageId>, // routing ID (key 13), if present
}

The presence or absence of an XID matters. (RID=A, XID=None) and (RID=A, XID=Some(5)) are two separate flows with independent counters. The same request may be seen with and without an XID at different points in the relay chain, before and after the RelaySwitch assigns one.

Source: frame.rs (FlowKey).

SeqAssigner

The SeqAssigner assigns monotonically increasing sequence numbers per FlowKey. It sits at the output stage, in the writer task and not at frame creation time. This ensures that the seq values reflect the actual order frames are written to the wire.

pub struct SeqAssigner {
    counters: HashMap<FlowKey, u64>,
}

For each flow, the counter starts at 0 and increments by 1 with each frame. There are no gaps. Non-flow frames (Hello, Heartbeat, RelayNotify, RelayState) are skipped, and the assigner leaves their seq at 0.

After a terminal frame (END or ERR) is assigned, the flow’s counter should be cleaned up via remove().

Source: frame.rs (SeqAssigner).

Reorder buffer

The ReorderBuffer sits at relay boundaries (specifically in the RelaySlave, on the socket → local path) and ensures frames are delivered in sequence order.

What an arriving frame does (delivered with those it releases, held, or refused) is decided by the proved model (formal/CapDAG/Bifaci/Flow.lean, Reorder.accept), generated into every mirror. The buffer keeps the frames; the model keeps the number expected next and the numbers held, and it is proved that frames are handed on in the order they were written (delivered_in_order), that none is lost (nothing_is_lost), and that once every frame has arrived every frame has been delivered (all_arrived_all_delivered).

pub struct ReorderBuffer {
    flows: HashMap<FlowKey, FlowState>,
    max_buffer_per_flow: usize,
}

struct FlowState {
    order: Reorder,              // the model's: expected seq + the seqs held
    buffer: HashMap<u64, Frame>, // the held frames themselves
}

When a frame arrives:

flowchart TD
    A["Frame arrives with seq"] --> B{seq == expected?}
    B -->|Yes| C["Deliver frame"]
    C --> D["Drain consecutive<br/>buffered frames"]
    D --> E["Advance expected_seq"]

    B -->|seq > expected| F["Buffer frame"]
    F --> G{Buffer size > max?}
    G -->|Yes| H["Protocol error"]
    G -->|No| I["Wait for gap fill"]

    B -->|seq < expected<br/>or duplicate| H

1. In order (seq == expected_seq): Delivered immediately, along with any consecutive frames already buffered. The expected_seq advances past all delivered frames.
2. Out of order (seq > expected_seq): Buffered. The frame waits until the gap is filled.
3. Stale or duplicate (seq < expected_seq, or seq already buffered): Protocol error. The frame is rejected.

If the buffer holds more than max_buffer_per_flow frames (default 64) for a single flow, that is also a protocol error, because the gap is too large and something has gone wrong.

Non-flow frames bypass the reorder buffer entirely and are returned immediately.

After a terminal frame (END or ERR) is delivered, cleanup_flow() removes the flow’s state from the buffer.

Source: frame.rs (ReorderBuffer).