Runtime della cartuccia — Documentazione di CapDAG

Il punto d’ingresso lato cartuccia: registrazione degli handler, rilevamento della modalità e ciclo principale di elaborazione dei frame.

CartridgeRuntime

CartridgeRuntime è l’unico modo supportato con cui una cartuccia comunica con l’host. Una cartuccia crea un runtime con il proprio manifest, registra gli handler per ogni capacità che fornisce e chiama run(). Il runtime gestisce tutto il resto: handshake, parsing dei frame, instradamento delle richieste, framing delle risposte e risposte agli heartbeat.

let manifest = build_manifest();
let mut runtime = CartridgeRuntime::with_manifest(manifest);

runtime.register_op("cap:in=...;out=...;generate", || {
    Box::new(MyGenerateOp::default())
});

runtime.run().unwrap();

La struct in sé è piccola:

pub struct CartridgeRuntime {
    handlers: HashMap<String, OpFactory>,   // cap URN → factory dell'handler
    manifest_data: Vec<u8>,                 // manifest codificato in JSON (inviato in HELLO)
    manifest: Option<CapManifest>,          // manifest analizzato (per la modalità CLI)
    limits: Limits,                         // limiti di protocollo negoziati
}

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

Costruzione del manifest

Il manifest indica all’host quali capacità fornisce questa cartuccia. Viene costruito con CapManifest:

let manifest = CapManifest::new(
    "mycartridge".to_string(),
    "1.0.0".to_string(),
    "Description of what this cartridge does".to_string(),
    vec![CapGroup::default_group(vec![identity_cap(), my_cap_1, my_cap_2])],
);

Ogni Cap nell’elenco ha:

Un cap URN costruito tramite CapUrn::from_string() o un builder, che definisce quali input la cap accetta, quali output produce e quali tag di operazione porta.
Un title e uno slug per la visualizzazione e per il nome del sottocomando CLI.
Argomenti (CapArg): ciascuno con un media URN, un flag di obbligatorietà e una o più sorgenti (ArgSource::Stdin, ArgSource::FilePath, ArgSource::Slot).

Il manifest deve sempre includere identity_cap(), la capacità di identità richiesta per la verifica dell’handshake (vedi Handshake). CartridgeRuntime::new() va in panic se manca CAP_IDENTITY (cap:effect=none).

Sorgente: capdag-rs/src/bifaci/manifest.rs (CapManifest); capdag-rs/src/cap/definition.rs (Cap, CapArg, ArgSource).

Registrazione degli handler

Due metodi registrano gli handler:

register_op(cap_urn, factory)

Registra una closure factory che crea una nuova istanza di Op<()> per ogni richiesta:

runtime.register_op("cap:in=...;out=...;describe", || {
    Box::new(DescribeOp::new(some_config))
});

La factory viene chiamata una volta per ogni REQ in arrivo. Questo permette agli handler di mantenere uno stato per richiesta.

register_op_type::<T>(cap_urn)

Registra un tipo che implementa Op<()> + Default. Le istanze vengono create tramite T::default():

runtime.register_op_type::<ExtractMetadataOp>("cap:in=...;out=...;extract;target=metadata");

Ricerca dell’handler

Quando arriva una REQ, il runtime trova l’handler confrontando il cap URN della richiesta con i cap URN registrati tramite is_dispatchable(candidate, request) (vedi Dispatch). Quando più handler corrispondono, il runtime li ordina per specificità, preferendo le corrispondenze esatte, poi i raffinamenti (più specifici di quanto richiesto), poi i fallback (più generici).

La stringa del cap URN usata in register_op deve poter essere analizzata da CapUrn::from_string(). Non deve essere identica carattere per carattere alla voce del manifest, ma deve produrre un dispatch equivalente.

Sorgente: cartridge_runtime.rs (register_op, register_op_type, find_handler).

Rilevamento della modalità

flowchart TD
    A["runtime.run()"] --> B{"Argomenti CLI?"}
    B -->|Nessun argomento| C["Modalità CBOR della cartuccia<br/>Legge/scrive frame binari<br/>su stdin/stdout"]
    B -->|Con argomenti| D{"Primo argomento?"}
    D -->|"manifest"| E["Stampa il manifest JSON"]
    D -->|"--help"| F["Stampa i sottocomandi"]
    D -->|corrispondenza di slug| G["Modalità CLI<br/>Analizza gli argomenti dalle definizioni delle cap<br/>Esegue l'handler, output RAW su stdout"]
    D -->|nessuna corrispondenza| H["Errore UnknownSubcommand"]

run() controlla gli argomenti della riga di comando del processo:

Nessun argomento: modalità CBOR della cartuccia. Il runtime legge frame binari da stdin e scrive su stdout. È così che l’host invoca le cartucce.
Qualsiasi argomento: modalità CLI. Il runtime analizza gli argomenti a partire dalle definizioni delle cap ed esegue direttamente l’handler corrispondente. L’output va su stdout.

Questo design a doppia modalità significa che un singolo binario funge sia da cartuccia CapDAG (quando viene avviato dall’host senza argomenti) sia da strumento CLI autonomo (quando viene eseguito direttamente da un utente).

La modalità CLI supporta:

il sottocomando manifest: stampa il manifest JSON.
il sottocomando <slug>: trova la cap il cui slug corrisponde, analizza gli argomenti a partire dalle definizioni CapArg della cap e invoca l’handler.
--help: elenca i sottocomandi disponibili.

Sorgente: cartridge_runtime.rs (run).

Ciclo principale in modalità cartuccia

Quando è in esecuzione in modalità CBOR della cartuccia, run() esegue questi passi:

Handshake: chiama handshake_accept() per scambiare i frame HELLO e negoziare i limiti. L’host avvia lo scambio; la cartuccia risponde con il proprio manifest.
Avvio del thread writer: un semplice thread del sistema operativo riceve frame da un canale std::sync::mpsc e li scrive su stdout. È l’unica cosa che tocca stdout; gli handler inviano attraverso un canale tokio che un task di inoltro svuota in esso.

2a. Avvio del thread reader (spawn_frame_reader): un secondo semplice thread del sistema operativo, con un piccolo runtime proprio, legge i frame da stdin. RISPONDE DA SÉ A UN HEARTBEAT, direttamente nella coda del thread writer, e passa ogni frame (heartbeat compresi, in ordine di arrivo) al ciclo principale. Nulla di ciò che sta tra l’arrivo di una sonda e la partenza della sua risposta viene eseguito sul runtime degli handler.

Ciclo di lettura: legge i frame da stdin in un ciclo.
- REQ: cerca un handler tramite cap_urn, risolvendo il pattern registrato che la servirà. Ammette la richiesta attraverso l’intera catena di pool della cap (vedi Pool di concorrenza più sotto); se un qualsiasi pool della catena è pieno, la accoda nella coda singleton della cap e invia un LOG level="queued" che indica la posizione in coda e il pool che blocca. Le richieste ammesse avviano un nuovo task handler (tokio::spawn) con il proprio InputPackage e il proprio OutputStream. Registra l’ID della richiesta perché i frame di continuazione possano essere instradati a questo task.
- Heartbeat: ha già risposto il thread reader, che ha applicato le eventuali desired_capacities portate dalla sonda (un pool sconosciuto fa rifiutare l’intero lotto con un ERR UNKNOWN_POOL al posto della risposta) e ha risposto con lo stesso ID e la mappa di stato pools aggiornata e obbligatoria. Il frame raggiunge questo ciclo solo perché esso compia un giro: una capacità aumentata ammette ciò che era in attesa.
- STREAM_START / CHUNK / STREAM_END / END: li instrada al task handler che possiede l’ID della richiesta. Questi frame alimentano l’InputPackage dell’handler.
- LOG / ERR (dalle risposte peer): li instrada al canale delle risposte peer del task handler.
- CloseStream: la chiusura del rubinetto (12.2 §CloseStream, 15.2 §Le esecuzioni si arrestano). Chiude il rubinetto o i rubinetti del feed dal vivo della richiesta (quello legato allo stream_id del frame, o tutti quando è assente) e fa continue: i feed terminano, l’handler svuota il proprio input e la richiesta termina naturalmente con END e output completi. Nulla viene annullato. Un CloseStream per una richiesta che non ha alcun feed aperto corrispondente viene registrato nel log e ignorato, mai trasformato in un’interruzione.
- Cancel: annulla sempre. L’attribuzione del frame (meta.code / meta.attribution_class / meta.message, 12.2 §Cancel) viene letta con Frame::cancel_reason(); un Cancel non attribuito produce CancelReason::unattributed, mai un errore. Chiude tutti i feed che la richiesta detiene, poi annulla in modo cooperativo: una richiesta in coda viene rimossa e riceve subito come risposta l’ERR terminale del motivo; un handler in corso ha il proprio input chiuso, chi attende il suo credito rilasciato, le sue chiamate peer annullate con lo stesso motivo, e il suo ERR terminale (il codice, la classe e il messaggio del motivo: CANCELLED/user, ABORTED_COLLATERAL con la classe del fallimento d’origine, ABORTED con quella dell’host, o CANCELLED/internal per un annullamento non attribuito) viene rinviato finché l’handler non termina, così nessuno stream resta orfano. Il runtime conserva il motivo per ogni richiesta annullata; non ricava mai le parole dell’ERR dalle circostanze.
- Policy: sostituisce la policy corrente del runtime (12.2 §Policy). Il network() di ogni handler legge il nuovo valore al controllo successivo, anche gli handler già in esecuzione. Un REQ prima del primo Policy è una violazione del protocollo: il runtime risponde con un ERR PROTOCOL_VIOLATION e non lo inoltra mai. L’unica eccezione è CAP_IDENTITY, la verifica di identità dell’host, che fa parte dell’handshake (12.3).
- EOF su stdin: interrompe il ciclo e chiude.
Chiusura: attende il completamento di tutti i task handler, poi rilascia il task writer.

Sorgente: cartridge_runtime.rs (metodo run).

Thread writer e reader

Il writer è un thread del sistema operativo che cicla su un std::sync::mpsc::Receiver<Frame>:

graph LR
    HA["Handler A"] --> CH["canale mpsc"]
    HB["Handler B"] --> CH
    HC["Handler C"] --> CH
    CH --> WT["Thread writer"] --> OUT["stdout"]

Tutti gli handler e il runtime stesso inviano frame attraverso questo canale. Il writer serializza ogni frame in CBOR, antepone la lunghezza di 4 byte e scrive su stdout.

Perché nessuno dei due è un task sul runtime della cartuccia. Un heartbeat chiede se la cartuccia c’è, e l’host concede dieci secondi per la risposta (14.1). Finché i frame venivano letti da un task del runtime, gestiti dal ciclo principale sullo stesso runtime e passati al writer da un terzo task su di esso, la risposta aspettava qualunque cosa stessero facendo gli handler: un handler che teneva occupati i thread del runtime (l’hashing di un intero download, un lungo documento renderizzato pagina dopo pagina) faceva dichiarare la propria cartuccia CARTRIDGE_UNHEALTHY nel mezzo di un lavoro corretto, e il lavoro andava perso. Il percorso della risposta ora non tocca alcun thread del runtime degli handler.

Che cosa dice quindi un heartbeat: il processo è vivo e sta leggendo il proprio input. Non dice che gli handler stanno avanzando; quello è compito del monitoraggio dell’attività e degli stalli della richiesta stessa (13.4).

spawn_blocking conta ancora per i frame PROPRI di un handler: i suoi frame di avanzamento e di keepalive vengono inviati da task sul runtime degli handler, e un handler che blocca ogni thread worker (per esempio un caricamento sincrono di un modello via FFI) non può inviarli, quindi scatta il timeout di attività del motore.

run_with_keepalive() risolve il problema spostando il lavoro bloccante su tokio::task::spawn_blocking (un pool di thread separato), liberando i worker di tokio così che il task writer possa svuotare i frame di keepalive. Vedi Avanzamento e logging.

Pool di concorrenza

L’unico concetto di capacità del runtime (bifaci::pools). Una cap È un pool di uno: il runtime materializza un pool singleton per ogni pattern di handler registrato (indicizzato dal cap URN canonico), ogni pool condiviso che il manifest dichiara (PoolDeclarations; le appartenenze devono risolversi in pattern registrati, e una discrepanza è un errore bloccante all’avvio), e il pool riservato all che contiene tutte le cap.

Ogni pool porta tre numeri: declared (il valore predefinito distribuito con il manifest), configured (il numero dell’operatore, consegnato tramite le sonde heartbeat e inizialmente uguale a declared) e l’opzionale available (l’autodichiarazione della cartuccia; l’assenza significa statico, il caso normale). Il limite effettivo di ammissione è min(configured, available), dove 0 significa illimitato ovunque.

Un dispatch viene ammesso attraverso l’intera CATENA della sua cap (singleton, pool dichiarati che la contengono, all) in modo atomico: ogni pool deve avere posto, altrimenti la richiesta attende in fila (i pool condivisi non hanno una fila propria). Ogni volta che qualcosa cambia (un handler termina, l’operatore modifica un limite o la cartuccia dichiara che cosa può servire), il runtime ammette la richiesta che ha atteso più a lungo tra quelle la cui intera catena ha posto, poi la successiva, finché nessuno in fila può procedere: in ordine di arrivo tra tutte le cap, mai con preferenze per una cap. pool_handle(name) dà a un handler un handle che scrive solo available (per esempio abbassando gpu a concorrenza 0 durante il caricamento di un modello); non tocca mai declared o configured, e alzarlo risveglia il ciclo di esecuzione così che una richiesta in fila parta subito anziché al frame successivo dell’host.

Chi viene ammesso, e quando, è una decisione del modello dimostrato (formal/CapDAG/Bifaci/Pools.lean, le leggi in Laws/Pools.lean): il runtime conserva i tre numeri che ogni pool porta sul wire e le richieste in coda stesse; il modello conserva il limite di ogni pool, quante richieste vi detengono uno slot e la fila. È dimostrato che nessun pool supera mai il proprio limite, che un rilascio annulla un’ammissione, che la richiesta ammessa successivamente è la prima in fila la cui catena ha posto (nextUp_is_first), che nessuno slot resta libero mentre attende una richiesta che potrebbe usarlo (admits_whenever_possible), e che una richiesta in arrivo non passa mai avanti: viene ammessa subito esattamente quando la sua catena ha posto e nessuno già in fila potrebbe procedere (arrive_admits_iff).

La mappa completa dello stato dei pool viaggia nei meta di HELLO e in ogni risposta all’heartbeat; l’host la incorpora nelle statistiche del roster, così l’ammissione del motore ed entrambi i client vedono gli stessi numeri che il runtime impone.

Trait Op

Gli handler implementano il trait Op<()> del crate ops:

#[async_trait]
pub trait Op<T>: Send + Sync {
    async fn perform(&self, dry: &mut DryContext, wet: &mut WetContext) -> OpResult<T>;
    fn metadata(&self) -> OpMetadata;
}

perform() esegue la logica dell’handler. Riceve un DryContext (flag di controllo, stato del dry-run) e un WetContext (un archivio chiave-valore tipizzato). Il WetContext contiene un Arc<Request> sotto la chiave "request".
metadata() restituisce informazioni descrittive sull’operazione.

flowchart LR
    REQ["Frame REQ"] --> FH["find_handler<br/>(is_dispatchable)"]
    FH --> F["OpFactory"]
    F --> OP["Istanza di Op"]
    OP --> P["perform()"]
    P -->|Ok| CL["close() → STREAM_END + END"]
    P -->|Err| ER["Frame ERR"]

Il runtime chiama dispatch_op(), che:

Crea una Request a partire dall’input, dall’output e dall’invocatore peer.
Inserisce la Request in un WetContext nuovo.
Chiama op.perform().
In caso di successo, chiude lo stream di output (invia STREAM_END se lo stream era stato avviato).
In caso di errore, il runtime invia un frame ERR per la richiesta.

Sorgente: cartridge_runtime.rs (dispatch_op); crate ops.

Oggetto Request

La struct Request raggruppa le tre cose di cui un handler ha bisogno:

pub struct Request {
    input: Mutex<Option<InputPackage>>,
    output: Arc<OutputStream>,
    peer: Arc<dyn PeerInvoker>,
}

take_input() → InputPackage: il raggruppamento di tutti gli stream degli argomenti di input. Può essere chiamato una sola volta; una seconda chiamata restituisce un errore. Questo previene i bug da doppio consumo.
output() → &OutputStream: per emettere dati di risposta, avanzamento e messaggi di log.
peer() → &dyn PeerInvoker: per chiamare le capacità di altre cartucce. Vedi Invocazione peer.
network() → NetworkAccess: la policy di rete corrente dell’host (12.2 §Policy). Un handle economico che si può clonare e passare al codice che apre le connessioni, il quale lo interroga prima di ogni chiamata: allow(Reach::ThirdParty) per un servizio esterno, Reach::OurServices per quelli di MachineFabric, Reach::Telemetry per l’analisi d’uso. Una portata che la policy non consente restituisce un NetworkRefused che si converte nel fallimento NETWORK_POLICY / user con la policy nominata, e la chiamata non viene fatta. L’handle legge il valore corrente del runtime, mai una copia presa al momento dell’inoltro, così un download lungo o un flusso che chiede prima di ogni chiamata si ferma alla successiva quando l’host passa offline.

Gli handler recuperano la Request dal WetContext:

let req = wet.get_arc::<Request>(WET_KEY_REQUEST).unwrap();
let input = req.take_input()?;
let streams = input.collect_streams().await?;
let data = require_stream(&streams, "media:document;ext=pdf")?;
let out = req.output();
out.start(false, None)?;
out.write(result_bytes).await?;

Sorgente: cartridge_runtime.rs (Request).

Equivalente Swift

Il CartridgeRuntime Swift in capdag-objc/Sources/Bifaci/CartridgeRuntime.swift segue lo stesso design:

Costruzione: CartridgeRuntime(manifest: manifestData).
Registrazione: closure o tipi Op registrati tramite la stringa del cap URN.
run() si blocca tramite DispatchSemaphore mentre un ciclo di eventi asincrono interno gestisce i frame.
I tipi di errore rispecchiano le varianti Rust (CartridgeRuntimeError con casi per handshake, noHandler, ioError, ecc.).

Differenze principali rispetto a Rust:

Modello di concorrenza: Swift usa la concorrenza strutturata (async/await con Task) invece di tokio. Il dispatch degli handler usa Task + DispatchSemaphore.wait() anziché tokio::spawn.
I/O: letture sincrone con FileHandle su stdin invece di I/O asincrono. Il modello bloccante basato su semaforo significa che il ciclo di eventi gira su un singolo thread.
Rischio di deadlock: poiché dispatchOp() racchiude Op.perform() in Task + DispatchSemaphore.wait(), annidare un altro Task + semaforo dentro perform() manda in deadlock l’esecutore asincrono. Questo vincolo non esiste nel modello tokio di Rust.

Sorgente: capdag-objc/Sources/Bifaci/CartridgeRuntime.swift.