Input and output

How handlers receive input data and emit output data via typed streams.

Writes are async and flow-controlled

Every data write is flow-controlled (one credit per CHUNK, see Flow control), so the handler API is asynchronous:

• OutputStream::write, emit_list_item, and emit_cbor are async fn; they await when the receiver’s window is exhausted. Call them with .await.
• blocking_write / blocking_emit_list_item serve non-async contexts (FFI threads, spawn_blocking closures). StreamSender (the detached emitter) keeps its sync signature and acquires credit blockingly.
• log() / progress() stay sync and are never flow-controlled; progress must flow even when the data window is exhausted (L14).
• start_unbounded(is_sequence, meta) opens a stream with no length promise (L16); its STREAM_END omits chunk_count. InputStream::is_unbounded() reports the flag on the consuming side, and the collect_* helpers refuse unbounded streams with a hard error, so consume them with recv().
• finish(progress, message) declares the request’s terminal status, carried in the END frame’s terminal metadata (L5). Do NOT emit a trailing 100% progress LOG, because the END metadata is the final progress event.

InputStream

An InputStream represents a single named input stream arriving from the host. Each stream carries a media_urn identifying the data type (e.g., "media:document;ext=pdf", "media:enc=utf-8") and yields decoded CBOR values extracted from CHUNK frames.

Handlers never see raw frames, sequence numbers, checksums, or stream IDs. The runtime’s demux layer decodes CHUNK payloads, verifies checksums, and delivers clean CBOR values through the stream.

pub struct InputStream {
    media_urn: String,
    rx: mpsc::UnboundedReceiver<Result<ciborium::Value, StreamError>>,
}

Source: capdag-rs/src/bifaci/cartridge_runtime.rs.

Receiving data

Three consumption patterns:

Streaming: process values one at a time:

let mut stream = /* from InputPackage */;
while let Some(item) = stream.recv().await {
    let value = item?;
    // process each CBOR value
}

Accumulate bytes: collect everything into a byte buffer:

let bytes: Vec<u8> = stream.collect_bytes().await?;

collect_bytes() extracts the inner bytes from each Value::Bytes or Value::Text chunk and concatenates them. Non-byte CBOR types are CBOR-encoded before appending. Only use this when you know the stream is finite; infinite streams block forever.

Single value: expect exactly one chunk:

let value: ciborium::Value = stream.collect_value().await?;

Returns the first value and ignores the rest. Returns StreamError::Closed if the stream is empty.

InputPackage

An InputPackage bundles all input streams for a single request. Handlers get it by calling req.take_input() (which can only be called once; a second call returns an error to prevent double-consumption).

pub struct InputPackage {
    rx: mpsc::UnboundedReceiver<Result<InputStream, StreamError>>,
    _demux_handle: Option<JoinHandle<()>>,
}

The package yields InputStream objects as STREAM_START frames arrive from the wire. It returns None after the END frame, signaling all arguments have been delivered.

Three ways to consume an InputPackage:

Collect each stream separately (most common):

let input = req.take_input()?;
let streams: Vec<(String, Vec<u8>)> = input.collect_streams().await?;
// streams = [("media:document;ext=pdf", <bytes>), ("media:enc=utf-8", <bytes>)]

Each stream’s bytes are accumulated independently. Use the stream lookup helpers (below) to find specific arguments by media URN.

Concatenate all streams:

let all_bytes: Vec<u8> = input.collect_all_bytes().await?;

Useful when a cap has only one input stream and you just want the bytes.

Iterate streams one at a time:

while let Some(stream_result) = input.recv().await {
    let stream = stream_result?;
    println!("Got stream: {}", stream.media_urn());
    let bytes = stream.collect_bytes().await?;
}

Source: cartridge_runtime.rs (InputPackage).

Stream lookup helpers

After collecting streams with collect_streams(), use these free functions to find specific arguments:

Matching uses MediaUrn::is_equivalent(): both URNs must have the exact same tag set (order-independent). This is an exact match, not subsumption or pattern matching. Both sides know the argument media URNs from the cap definition, so equivalence is the right test.

let streams = input.collect_streams().await?;
let pdf_data = require_stream(&streams, "media:document;ext=pdf")?;
let prompt = find_stream_str(&streams, "media:enc=utf-8");

The media_urn parameter must be the full media URN from the cap’s argument definition.

Source: cartridge_runtime.rs (find_stream, require_stream).

Argument roles: the main input

Every cap argument has a role, and the contract that assigns it is the canonical arg-role model every front-end (the capdag CLI, the planner’s step-requirement payloads, UIs) presents:

• Main input. An arg is the cap’s main input iff it declares a stdin source (CapArg::is_main_input); validation RULE3 and RULE15 guarantee that a cap whose in= is not media:void has exactly one, and that its stdin URN is tagged-URN-equivalent to in=. The main input is the value piped in on stdin, like a Unix command’s stdin. Its declared URN may differ from the stdin URN (e.g. a file-path arg whose piped content is a pdf-stream); the stdin URN, not the arg’s own, is in=.
• Identity. An arg is identified by its media_urn, unique within the cap. Bindings, run-time values and diagnostics all name an arg by it.
• Name. Every arg has a name, unique within the cap (RULE17): the label a machine selects it by, and the spelling the CLI accepts in --arg <name>=. It identifies nothing, and nothing about the URN can be read from it.
• Accepted type. What a value bound to the arg must conform to (CapArg::accepted_urn()): the stdin source URN for the main input; accepts when the definition gives one; the arg’s own media_urn otherwise.
• Stream identity. The runtime demuxes each arg’s input stream by CapArg::stream_urn(): the arg’s stdin source URN when it declares one, otherwise its media_urn. The engine relabels a forwarded stream to that URN after auditing what it carries, so an op looks its args up by the URNs its own definition declares, whatever produced the value.
• Required options. Non-main args with required: true and no default_value; an invocation must supply each of them.
• Options. The remaining non-main args, carrying their default_value.

Every arg, main or not, may be supplied by another cap’s output, by a constant, or by a run-time value (Machine notation). “Option” describes how an arg is offered on a command line, not where its value may come from.

Consumers of the model: the planner marks the main input on every step argument (ArgumentInfo.is_main_input), and the CLI structures per-cap help and missing-argument errors around this three-way split (see CapDAG CLI reference §”Cap interface presentation”).

Reference media and transport resolution

Some inputs arrive by reference: the value on the wire is not the data, it is a small pointer the runtime resolves into the data. The contract is:

• The arg is declared with a reference media URN (the slot URN); its stdin source carries the content media URN; the cap URN’s in= equals the stdin URN. CapArg::is_main_input already matches on the stdin source, so a reference-fed main input is structurally identical to a directly-fed one. RULE3/RULE15 apply unchanged.
• The runtime’s transport-resolution layer, not the op, recognizes the reference URN on an incoming stream, consumes the reference value, resolves it, and delivers the resolved content to the op as an ordinary InputStream labeled with the content URN. The op is transport-blind: it cannot tell a resolved reference from directly piped content, and must never try.
• On platforms where the cartridge cannot reach the referenced resource (sandboxing, device permissions), the host resolves instead and streams the content over the wire as a normal input stream. Same cap, same plan, different resolver: a deployment detail, never a protocol one.

Two reference families exist:

File paths (bounded references)

Reference URN family: media:file-path (any URN it accepts). The reference value is a path string or, for is_sequence args, a newline-separated list of paths/globs. Resolution reads the file(s) and delivers the bytes as a bounded stream (scalar) or one item per file (sequence). A missing file, a failed glob, or a multi-file resolution into a scalar arg is a hard error. See FilePathContext in cartridge_runtime.rs.

Live feeds (unbounded references)

Reference URN family: any URN carrying the live marker tag: media:audio;live;microphone, media:image;live;webcam, media:live;synthetic (the built-in deterministic test feed). The reference value is a selector record (JSON):

Resolution opens the device via the built-in capture dispatch (capdag::capture::open, a CLOSED, compile-time set of backends matched by reference-URN family in plain code, exactly as file paths are read by plain code; capture is transport, not a capability, and there is no registry and nothing to register) and delivers an unbounded sequence stream of items labeled with the arg’s stdin content URN. Each item carries per-item metadata (pts_us, capture_ts_us, seq, and gap markers; see Streaming §Live feeds). An unregistered reference URN, an unparseable selector, or a device the provider cannot open is a hard error, never a silent empty stream. Selector parsing is strict at every nesting level: an unknown field (in the selector or inside stop) is rejected (a misspelled stop condition silently ignored would run an unbounded feed the caller meant to bound). An empty value is the all-defaults selector; its canonical spelling is {} (clients serialize selectors with defaults omitted and keys sorted, so equal selections are byte-equal). Family membership is the encapsulated is_live_feed media-URN predicate, present in all five implementations.

The backends: media:live;synthetic, a deterministic clock source (selector params: items, interval_ms, item_bytes, ring, plus overrun injection) used by the shared test range and available as a fixture feed in every build; and the hardware backends (microphone via alsa/avfoundation, webcam via v4l2/avfoundation) behind CapDAG’s capture cargo feature, enabled by the capture-capable builds (the CLI, the engine, audiocartridge, videocartridge). A reference outside the known families, or a hardware family in a build without the feature, is a hard, named error.

Which runtime resolves is a deployment fact, not a protocol one. The cartridge runtime resolves when the consumer is an op (the linear-machine case: the first consuming cap’s process opens the device). The HOST resolves when the cartridge cannot reach the device (sandboxing), and, the same rule from the other side, when the host itself is the consumer: a live source feeding a ForEach region directly is iterated by the ENGINE, so the engine opens the feed via the same dispatch and dispatches one body per delivered item while the capture runs. The engine registers the tap so a run stop closes it (close-tap → the feed ends → in-flight bodies drain → a stopped run is a valid run, 15.2 §Runs stop).

Backpressure and delivery. A live feed’s delivery channel into the op is bounded. The chain is: wire credit stalls the op’s output → the op stops consuming input → the bounded feed channel fills → the provider’s feeder stops draining the capture ring → the ring fills → the capture edge applies the overrun policy. Loss can occur ONLY at the capture edge, only under the declared policy, and always counted (§Overrun). This is the live analog of L10’s deadlock-freedom: every stage has a full-state behavior and none of them is “buffer forever”.

Ending. A feed ends when a stop condition fires, when the request is stopped (a CloseStream frame: the runtime closes the request’s live feeds and lets the pipeline drain; see Execution §Runs stop), or when the request is cancelled. The feed’s end is an ordinary input-stream end: the op observes it exactly as it observes a file stream ending. A cap that reads a feed declares it: streaming = true on its main input (15.2 §Streaming contracts), which is what lets the executor wire it live to an unbounded producer at all.

Response media labeling: the effect contract

The media URN a response STREAM_START carries is derived by the runtime, not chosen by the op. derive_response_media(cap_urn) (Rust; the same function exists in every runtime port) computes it from the cap’s declared effect over its declared main input; this is the label every engine-fed input stream carries:

This is CapUrn::apply_to_runtime_input_media over the declared input, the same inference the planner used to refine downstream types at plan time, so a runtime-labeled response satisfies the engine’s effect audit by construction. A cap whose label cannot be derived (unparseable URN, inapplicable effect) fails the request at dispatch; there is no fallback label.

The engine audits every emission at receipt (ResponseAudit, backed by the single predicate CapUrn::is_conformant_runtime_output): before a response STREAM_START is forwarded to the next step, persisted, or collected, its media URN must satisfy the producing cap’s declared effect contract. The condition is per-effect:

• effect=none / effect=patch compute an exact output type: the emission must be tag-equivalent to the inference. A more specific emission is still a violation: the effect promises the type is fully determined by the input.
• effect=declared promises only the declared out=: the emission must conform_to it. More specific is legal; more generic breaks every downstream refinement and fails.

A violation is a hard failure attributed internal (ExecutionError::EffectContractViolation), naming the cap URN, the declared effect, the runtime input, and expected vs actual, raised at the frame boundary, before the pipelined forward relabels the stream to the downstream declared arg URN, so the relabel is a verified assertion rather than a mask over a lying cap.

Outputs

A cap declares one or more outputs (none when out= is media:void). Each is identified by its media_urn, has a name (RULE19), and is optional or required. One is the main output: the one whose URN is the cap URN’s out=.

let fit = req.output();                                           // the main output
fit.start(false, None)?;
fit.write(b"the main output").await?;

let residuals = req.output_for("media:fmt=json;list;record;residual")?;
residuals.start(true, None)?;
residuals.emit_list_item(&row, None).await?;

req.output() is the main output’s stream. req.output_for(urn) is the stream of the output that URN identifies (tag-equivalence, as with input lookup); a URN the cap does not declare is RuntimeError::UnknownOutput. Each is an OutputStream (below) with its own STREAM_START, credit and STREAM_END.

The main output’s stream names no output: it is the cap’s answer. The stream of every other output names the output it is (frame key 21, that output’s media_urn) beside the media URN of what it carries. What an output carries, and how the engine audits it:

When the handler returns Ok(()):

• a required output the handler never wrote to is opened and closed by the runtime as an empty value (an empty sequence when it is is_sequence);
• an optional output the handler never wrote to opens no stream: it did not fire, and its consumers see it as absent (Machine notation).

The engine refuses, as ExecutionError::OutputContractViolation, a main-output stream from a cap that declares no output, a stream that names an output the cap does not declare or names its main output, a second stream for one output, an output other than the main one carrying a type that is not what it declares, and a request that ends without a stream for a required output.

OutputStream

OutputStream is the handler’s interface for emitting response data, progress, and log messages. It manages STREAM_START/CHUNK/STREAM_END framing automatically: handlers write data, and the stream handles the rest.

A stream holds its own stream_id, the media URN its STREAM_START carries, the output it is when it is a response stream (an argument stream of a peer call names none), its chunk counters, its flow-control window and its write-coalescing buffer.

Source: cartridge_runtime.rs (OutputStream).

Opening a stream

A handler opens a stream before writing to it, and says which kind it is:

req.output().start(false, None)?;     // one value, written in pieces
req.output().start(true, None)?;      // a sequence, one item at a time

start(is_sequence, meta) sends STREAM_START. is_sequence is the stream’s cardinality mode and must be the one the output declares; the engine audits it. meta is the stream’s metadata, forwarded to consumers. start_unbounded(is_sequence, meta) opens a stream that makes no length promise (a feed). Writing to a stream that was not started, starting one twice, and writing in the other mode than it was started in are each an error.

Emitting data

Raw bytes: auto-chunked at max_chunk boundaries:

req.output().write(b"result data here").await?;

Small writes COALESCE: they accumulate in a per-stream buffer and ship as one CHUNK when the batch reaches 4096 bytes or the oldest buffered byte is 20 ms old at the next write (COALESCE_MAX_BYTES / COALESCE_MAX_AGE). Chunk boundaries on a scalar stream are non-semantic, so this is invisible to receivers, and a per-token emitter produces batches instead of one frame per token. A batch that is due (and any write larger than the threshold) splits at max_chunk boundaries (default 256 KB); each piece is wrapped as Value::Bytes and sent as a CHUNK frame with checksum, one credit per frame. flush() (and blocking_flush) force the buffered tail out for explicit mid-stream latency barriers; close() always flushes it.

Single CBOR value:

req.output().emit_cbor(&ciborium::Value::Text("hello".to_string())).await?;

Handles type-specific chunking: Bytes values coalesce exactly like write (pure byte-stream continuation on a scalar stream); Text is split at max_chunk boundaries (respecting UTF-8 character boundaries). Array elements are sent one per chunk. Map entries are sent as [key, value] pairs. Scalar types are sent as a single chunk. Every non-Bytes emission is an ORDERING BARRIER: it flushes the coalescing buffer first, so nothing overtakes buffered bytes. Sequence items (emit_list_item) are never coalesced; their chunk runs delimit items.

List item (for streaming list output):

for item in items {
    req.output().emit_list_item(&item, None).await?;
}

The second argument is the item’s own metadata, carried on its chunk.

Each call CBOR-encodes the value and sends the raw bytes as chunk payloads. The receiver concatenates chunk payloads to produce an RFC 8742 CBOR sequence: one self-delimiting CBOR value per emit_list_item call. This differs from emit_cbor, which re-wraps each piece as a separate CBOR value.

Progress and logging:

req.output().progress(0.5, "Processing page 3 of 6");
req.output().log("info", "Found 12 images");

These emit LOG frames that travel alongside data frames. See Progress and logging.

Close:

req.output().close().await?;

Flushes any coalesced tail bytes, then sends STREAM_END with the total (coalesced) chunk count; flushing before the count is promised is what makes coalescing lossless. close is async because the tail flush may wait on credit; blocking_close() is the FFI-thread counterpart. Idempotent: calling it twice is safe.

What the runtime does for the handler

sequenceDiagram
    participant H as Handler
    participant OS as OutputStream
    participant W as Wire

    H->>OS: start(is_sequence, meta)
    OS->>W: STREAM_START
    H->>OS: write(data) or emit_cbor(value)
    OS->>W: CHUNK (data)

    H->>OS: write(more)
    OS->>W: CHUNK (more)

    Note over H: handler returns Ok(())
    Note over OS: runtime calls close()
    OS->>W: STREAM_END
    Note over OS: runtime sends
    OS->>W: END

• STREAM_START: sent by start(), which the handler calls. On the stream of an output other than the main one it names that output (frame key 21).
• STREAM_END: sent by close(). The runtime closes every stream the handler opened once the handler returns Ok(()), so a handler need not.
• An output the handler never opened: see Outputs. A required one is opened and closed by the runtime as an empty value; an optional one sends nothing.
• END: sent by the runtime after the handler returns.
• On error: the runtime sends an ERR frame instead of END, and closes nothing: what the handler had written stays as it was, and the request has failed.

A cartridge run from its own command line

A cartridge invoked directly (<cartridge> <alias> ...) has one standard output, and it carries the cap’s main output: the stream that names no output. Another output that fires is not printed: the cartridge says so on standard error, naming the output. A machine is what routes a cap’s other outputs.

CBOR encoding conventions

Arguments and results flow as CBOR values in CHUNK payloads. The encoding conventions:

• Text arguments (prompts, file paths, configuration strings): sent as Value::Bytes containing UTF-8 encoded text, not JSON-quoted. A prompt string "hello world" becomes the raw bytes hello world, not "hello world".
• Binary data (images, PDFs, model weights): sent as Value::Bytes containing raw binary data.
• JSON structures (model specs, configuration objects): sent as Value::Bytes containing the JSON string in UTF-8.
• Numeric arguments: sent as Value::Bytes containing the string representation in UTF-8.

This matches the json_value_to_bytes convention used in the planner module (see Planner): strings become raw UTF-8 bytes (no JSON quoting), everything else becomes JSON-encoded bytes.

Swift equivalent

The Swift OutputStream and InputStream in capdag-objc/Sources/Bifaci/CartridgeRuntime.swift follow the same semantics with platform-appropriate types:

• Data instead of Vec<u8>, CBOR values instead of ciborium::Value.
• Methods throw instead of returning Result.
• start(isSequence:meta:), write(_:), emitCbor(_:), emitListItem(_:meta:), close(), as in Rust; blockingClose() for a thread that cannot await.
• log(level:attributionClass:message:argUrn:) and progress(_:message:) for diagnostics; finish(progress:message:) declares the request’s terminal status.
• runWithKeepalive() uses a background Task instead of tokio::task::spawn_blocking.
• CborRequest.output() is the main output; output(for:) finds another declared output by its media URN, as Request::output_for does.

The stream lifecycle is identical: STREAM_START when start is called, STREAM_END on close, END on handler completion; when the handler returns, a required output it never wrote to is sent empty and an optional one is not sent.