Pattern degli handler

Come implementare gli handler delle richieste: il trait Op, l’estrazione degli argomenti, l’emissione dell’output e i pattern più comuni.

Implementazione del trait Op

Rust

Ogni handler implementa 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()
    }
}

Sorgente: handler di pdfcartridge, txtcartridge, ggufcartridge, candlecartridge.

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()
    }
}

Sorgente: handler di mlxcartridge.

Estrazione degli argomenti

flowchart LR
    WET["WetContext"] -->|get_arc| REQ["Request"]
    REQ -->|take_input| IP["InputPackage"]
    IP -->|collect_streams| ST["Vec<(media_urn, bytes)>"]
    ST -->|require_stream| DATA["Byte dell'argomento"]
    ST -->|find_stream_str| OPT["Testo facoltativo"]

    REQ -->|output| OS["OutputStream"]
    OS -->|emit_cbor| OUT["Risposta"]
    OS -->|progress| LOG["Frame LOG"]

Raccogliere gli stream

Il pattern standard per ottenere gli argomenti dall’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() attende che arrivino tutti gli stream di input e accumula separatamente i byte di ciascuno stream. Ogni tupla è (media_urn, bytes); il media URN identifica di quale argomento si tratta.

Cercare gli argomenti

Dopo aver raccolto gli stream, usa le funzioni di ricerca per trovare argomenti specifici:

// Argomento binario obbligatorio: fallisce se manca
let pdf_data = require_stream(&streams, "media:document;ext=pdf")?;

// Argomento testuale obbligatorio: fallisce se manca o non è UTF-8 valido
let prompt = require_stream_str(&streams, "media:enc=utf-8")?;

// Argomento binario facoltativo: restituisce None se manca
let image_data = find_stream(&streams, "media:ext=png;image");

// Argomento testuale facoltativo: restituisce None se manca
let model_override = find_stream_str(&streams, "media:model-spec;enc=utf-8");

La corrispondenza usa MediaUrn::is_equivalent(): un confronto esatto degli insiemi di tag, non la sussunzione. Il media URN deve corrispondere esattamente alla definizione dell’argomento della cap.

Analizzare gli argomenti numerici

Gli argomenti numerici arrivano come rappresentazioni in stringa UTF-8, non come numeri CBOR. La convenzione:

let max_tokens: u32 = find_stream_str(&streams, "media:max-tokens;enc=utf-8;numeric")
    .and_then(|s| s.parse().ok())
    .unwrap_or(50);  // valore predefinito se manca o non è analizzabile

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

Questo segue la convenzione di json_value_to_bytes (vedi Planner): le stringhe diventano byte UTF-8 grezzi, i numeri sono codificati come stringhe.

Emissione dell’output

Valore singolo

Per le cap che producono un unico risultato (metadati, informazioni sul modello, embedding):

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

start invia STREAM_START; il runtime invia STREAM_END ed END dopo che l’handler è tornato.

Lista in streaming

Per le cap che producono più elementi (disbind, estrazione delle pagine):

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?;
}

Ogni chiamata a emit_list_item() produce un elemento in una sequenza CBOR RFC 8742. Il destinatario concatena i byte CBOR grezzi.

Output testuale

I risultati testuali sono codificati come byte CBOR (UTF-8), non come stringhe di testo CBOR:

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

Questo è coerente con la convenzione per cui tutti i dati del payload fluiscono come Value::Bytes.

Più output

Una cap che dichiara più di un output scrive ciascuno sul proprio stream, scelto con il media URN dell’output esattamente come lo scrive la definizione:

// fit: l'output principale (il `out=` del cap URN)
req.output().start(false, None)?;
req.output().emit_cbor(&fit).await?;
// residuals: un secondo output dichiarato
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?;
}

Lo stream dell’output principale non nomina alcun output; ogni altro stream nomina il proprio, così il motore sa quale è quale anche quando due output portano valori dello stesso tipo.

output_for con un URN che la cap non dichiara è RuntimeError::UnknownOutput: un difetto della cartridge, non qualcosa da cui riprendersi.

Un output che può non scattare

Un output dichiarato optional scatta solo se l’handler vi scrive. Un router scrive ogni elemento su uno dei suoi output e lascia stare l’altro:

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

Tutto ciò che sta a valle dell’output che non è scattato viene saltato per quell’elemento. Un output obbligatorio che l’handler non scrive mai non viene saltato: il runtime lo chiude come valore vuoto. Dichiarare optional nella definizione quando “niente” è una risposta reale.

Pattern di emissione dell’avanzamento

Tappe standard (cartucce ML)

Tappe non ML (cartucce di contenuto)

Queste sono convenzioni. Il protocollo non impone alcuna disposizione particolare.

Registrazione per più tipi

Quando lo stesso handler serve più tipi di media, registralo una volta per ciascun tipo:

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));
}

L’handler usa una ricerca di ripiego per trovare il media URN, qualunque sia, che corrisponde allo stream in arrivo:

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()))
}

Il dispatch dei cap URN garantisce che arrivi soltanto il tipo di media corretto, quindi esattamente una delle chiamate a find_stream troverà una corrispondenza.

Sorgente: txtcartridge/src/main.rs.

Gestione degli errori negli handler

• Restituisci OpError::ExecutionFailed(message) per gli errori nella logica dell’handler.
• Converti RuntimeError e StreamError in OpError al confine dell’handler.
• Non intercettare e ingoiare mai gli errori; fallisci in modo netto per far emergere i problemi. Un errore soppresso diventa una corruzione silenziosa dei dati o una richiesta bloccata.
• Il runtime converte automaticamente gli errori degli handler in frame ERR e li invia all’engine.

// Corretto: propagare gli errori
let data = require_stream(&streams, "media:document;ext=pdf")
    .map_err(|e| OpError::ExecutionFailed(e.to_string()))?;

// Sbagliato: ingoiare gli errori (NON FARLO)
// let data = find_stream(&streams, "media:document;ext=pdf").unwrap_or_default();