Handler patterns — CapDAG documentation

How to implement request handlers: the Op trait, argument extraction, output emission, and common patterns.

Op trait implementation

Rust

Every handler implements Op<()>:

struct ExtractMetadataOp;

#[async_trait]
impl Op<()> for ExtractMetadataOp {
    async fn perform(&self, _dry: &mut DryContext, wet: &mut WetContext) -> OpResult<()> {
        let req = wet.get_arc::<Request>(WET_KEY_REQUEST)
            .ok_or_else(|| OpError::ExecutionFailed("Missing request".to_string()))?;

        let input = req.take_input()?;
        let streams = input.collect_streams().await
            .map_err(|e| OpError::ExecutionFailed(e.to_string()))?;

        let pdf_data = require_stream(&streams, "media:document;ext=pdf")
            .map_err(|e| OpError::ExecutionFailed(e.to_string()))?;

        let metadata = extract_metadata(pdf_data)?;
        let result = serde_json::to_vec(&metadata)?;
        let out = req.output();
        out.start(false, None)
            .map_err(|e| OpError::ExecutionFailed(e.to_string()))?;
        out.emit_cbor(&ciborium::Value::Bytes(result))
            .await
            .map_err(|e| OpError::ExecutionFailed(e.to_string()))?;

        Ok(())
    }

    fn metadata(&self) -> OpMetadata {
        OpMetadata::builder("ExtractMetadata").build()
    }
}

Source: pdfcartridge, txtcartridge, ggufcartridge, candlecartridge handlers.

Swift

struct ExtractMetadataOp: Op, Sendable {
    typealias Output = Void

    func perform(dry: DryContext, wet: WetContext) async throws {
        let req = try wet.getRequest()
        let input = try req.takeInput()
        let streams = try await input.collectStreams()

        let pdfData = try requireStream(streams, mediaUrn: "media:document;ext=pdf")
        let metadata = extractMetadata(pdfData)
        let result = try JSONEncoder().encode(metadata)
        try req.output().emit(chunk: result)
    }

    func metadata() -> OpMetadata {
        OpMetadata.builder("ExtractMetadata").build()
    }
}

Source: mlxcartridge handlers.

Argument extraction

flowchart LR
    WET["WetContext"] -->|get_arc| REQ["Request"]
    REQ -->|take_input| IP["InputPackage"]
    IP -->|collect_streams| ST["Vec&lt;(media_urn, bytes)&gt;"]
    ST -->|require_stream| DATA["Argument bytes"]
    ST -->|find_stream_str| OPT["Optional text"]

    REQ -->|output| OS["OutputStream"]
    OS -->|emit_cbor| OUT["Response"]
    OS -->|progress| LOG["LOG frame"]

Collecting streams

The standard pattern for getting arguments from the input:

let req = wet.get_arc::<Request>(WET_KEY_REQUEST).unwrap();
let input = req.take_input()?;
let streams = input.collect_streams().await?;
// streams: Vec<(String, Vec<u8>)> — [(media_urn, bytes), ...]

collect_streams() waits for all input streams to arrive and accumulates each stream’s bytes separately. Each tuple is (media_urn, bytes); the media URN identifies which argument it is.

Looking up arguments

After collecting streams, use the lookup helpers to find specific arguments:

// Required binary argument — fails if missing
let pdf_data = require_stream(&streams, "media:document;ext=pdf")?;

// Required text argument — fails if missing or not valid UTF-8
let prompt = require_stream_str(&streams, "media:enc=utf-8")?;

// Optional binary argument — returns None if missing
let image_data = find_stream(&streams, "media:ext=png;image");

// Optional text argument — returns None if missing
let model_override = find_stream_str(&streams, "media:model-spec;enc=utf-8");

Matching uses MediaUrn::is_equivalent(): exact tag-set comparison, not subsumption. The media URN must match the cap’s argument definition exactly.

Parsing numeric arguments

Numeric arguments arrive as UTF-8 string representations, not as CBOR numbers. The convention:

let max_tokens: u32 = find_stream_str(&streams, "media:max-tokens;enc=utf-8;numeric")
    .and_then(|s| s.parse().ok())
    .unwrap_or(50);  // default if missing or unparseable

let temperature: f32 = find_stream_str(&streams, "media:temperature;enc=utf-8;numeric")
    .and_then(|s| s.parse().ok())
    .unwrap_or(0.7);

This follows the json_value_to_bytes convention (see Planner): strings become raw UTF-8 bytes, numbers are string-encoded.

Output emission

Single value

For caps that produce a single result (metadata, model info, embeddings):

let result_bytes = serde_json::to_vec(&response)?;
req.output().start(false, None)?;
req.output().emit_cbor(&ciborium::Value::Bytes(result_bytes)).await?;

start sends STREAM_START; the runtime sends STREAM_END and END after the handler returns.

Streaming list

For caps that produce multiple items (disbind, page extraction):

req.output().start(true, None)?;
for (i, page) in pages.iter().enumerate() {
    req.output().progress(i as f32 / pages.len() as f32, &format!("Page {}", i + 1));
    req.output().emit_list_item(&ciborium::Value::Bytes(page.to_vec()), None).await?;
}

Each emit_list_item() call produces one item in an RFC 8742 CBOR sequence. The receiver concatenates the raw CBOR bytes.

Text output

Text results are encoded as CBOR bytes (UTF-8), not as CBOR text strings:

req.output().start(false, None)?;
req.output().emit_cbor(&ciborium::Value::Bytes(text.into_bytes())).await?;

This matches the convention that all payload data flows as Value::Bytes.

Several outputs

A cap that declares more than one output writes each on its own stream, selected by the output’s media URN exactly as the definition spells it:

// fit: the main output (the cap URN's `out=`)
req.output().start(false, None)?;
req.output().emit_cbor(&fit).await?;
// residuals: a second, declared output
let residuals = req.output_for("media:fmt=json;list;record;residual;time-series")?;
residuals.start(true, None)?;
for point in &points {
    residuals.emit_list_item(point, None).await?;
}

The main output’s stream names no output; each other stream names its own, so the engine knows which is which even when two outputs carry values of one type.

output_for with a URN the cap does not declare is RuntimeError::UnknownOutput: a defect in the cartridge, not something to recover from.

An output that may not fire

An output declared optional fires only if the handler writes to it. A router writes each item to one of its outputs and leaves the other alone:

let out = if is_valid(&row) { req.output_for(KEPT)? } else { req.output_for(DROPPED)? };
out.start(false, None)?;
out.write(&row_bytes).await?;

Everything downstream of the output that did not fire is skipped for that item. A required output the handler never writes is not skipped: the runtime closes it as an empty value. Say optional in the definition when “nothing” is a real answer.

Progress emission patterns

Standard milestones (ML cartridges)

Progress	Activity
0.00	Starting (before peer call).
[0.00, 0.25]	Model download progress (mapped from peer call).
0.25	Loading model into memory.
0.35	Model loaded, starting inference.
[0.35, 0.95]	Inference progress (per-token or per-step).
0.95	Complete.

Non-ML milestones (content cartridges)

Progress	Activity
0.05	Starting processing.
0.50	Midway (e.g., pages processed).
0.90	Encoding output.

These are conventions. The protocol does not enforce any particular layout.

Multi-type registration

When the same handler serves multiple media types, register it once per type:

for media_urn in &["media:enc=utf-8", "media:rst;enc=utf-8", "media:log;enc=utf-8", "media:enc=utf-8;ext=md"] {
    runtime.register_op(&generate_thumbnail_urn(media_urn), || Box::new(ThumbnailOp));
    runtime.register_op(&extract_metadata_urn(media_urn), || Box::new(MetadataOp));
}

The handler uses a fallback lookup to find whichever media URN matches the incoming stream:

fn require_text_content(streams: &[(String, Vec<u8>)]) -> Result<&[u8], StreamError> {
    for urn in &["media:enc=utf-8", "media:rst;enc=utf-8", "media:log;enc=utf-8", "media:enc=utf-8;ext=md"] {
        if let Some(bytes) = find_stream(streams, urn) {
            return Ok(bytes);
        }
    }
    Err(StreamError::Protocol("No recognized text stream found".to_string()))
}

Cap URN dispatch ensures only the correct media type arrives, so exactly one of the find_stream calls will match.

Source: txtcartridge/src/main.rs.

Error handling in handlers

Return OpError::ExecutionFailed(message) for handler logic errors.
Map RuntimeError and StreamError to OpError at the handler boundary.
Never catch and swallow errors; fail hard to expose problems. A suppressed error becomes a silent data corruption or a hung request.
The runtime converts handler errors to ERR frames automatically and sends them to the engine.

// Good: propagate errors
let data = require_stream(&streams, "media:document;ext=pdf")
    .map_err(|e| OpError::ExecutionFailed(e.to_string()))?;

// Bad: swallow errors (DON'T DO THIS)
// let data = find_stream(&streams, "media:document;ext=pdf").unwrap_or_default();