Avanzamento e logging

Come le cartucce emettono aggiornamenti di avanzamento e messaggi di log, e come questi fluiscono attraverso la catena di esecuzione.

Frame LOG

I frame LOG sono il modo in cui le cartucce comunicano lo stato all’engine e alla UI durante la gestione di una richiesta. Non sono terminali: un handler può emettere frame LOG in qualsiasi momento tra il primo STREAM_START e l’END finale. I frame LOG viaggiano accanto ai frame di dati sullo stesso canale della richiesta e si intercalano liberamente con i frame CHUNK.

Ogni frame LOG porta un testo level e un testo message non vuoti nella chiave di metadati 5. I frame LOG non di avanzamento portano anche un attribution_class obbligatorio (input, resource, environment o internal) e possono portare un arg_urn non vuoto dichiarato dalla sorgente. I frame LOG di avanzamento portano invece un float progress e non portano mai un’attribuzione diagnostica.

I frame LOG non influiscono sullo stream di dati. L’avanzamento alimenta lo stato di avanzamento in tempo reale. Le diagnostiche non di avanzamento vengono inoltrate in tempo reale con il token stabile del passo di esecuzione e l’ordinale facoltativo del corpo; i client conservano una finestra limitata in memoria. Non vengono rese persistenti come cronologia strutturata delle esecuzioni. I riceventi non deducono mai una classe o un’attribuzione di argomento mancante, e un’attribuzione malformata è un errore di protocollo.

Sorgente: capdag-rs/src/bifaci/frame.rs (Frame::log, Frame::progress).

Livelli di log

Livelli standard:

I livelli personalizzati sono consentiti. L’engine li lascia passare senza interpretarli.

Frame di avanzamento

I frame di avanzamento sono frame LOG con level = "progress" e un float progress aggiuntivo (0.0–1.0) nella mappa meta. L’engine usa questo valore per aggiornare l’avanzamento del task nel database e nella UI.

// Costruzione di un frame di avanzamento
Frame::progress(request_id, 0.5, "Processing page 3 of 6");
// Produce meta: {"level": "progress", "message": "Processing page 3 of 6", "progress": 0.5}

Il metodo log_progress() di Frame restituisce Some(f32) solo quando il livello è "progress". Un frame LOG con un livello diverso e un campo progress in meta restituisce None, perché il controllo del livello è rigoroso.

I valori di avanzamento non funzionano a cricchetto. Un handler può emettere 0.8 seguito da 0.3. Non c’è alcuna imposizione di monotonicità nel protocollo o nel database, perché alcune operazioni tornano effettivamente su stati precedenti (ad esempio la codifica multi-passata).

Sorgente: frame.rs (Frame::progress, log_progress).

Emettere l’avanzamento

Da un handler, usa i metodi di OutputStream:

// Aggiornamento di avanzamento con valore e messaggio
req.output().progress(0.5, "Halfway through inference");

// Diagnostica a livello di passo
req.output().log(
    "info",
    AttributionClass::Internal,
    "Found 12 images in document",
);

// Diagnostica attribuita dalla sua sorgente a un argomento della cap
req.output().log_for_argument(
    "warn",
    AttributionClass::Resource,
    "Model cache is under memory pressure",
    MEDIA_MODEL_SPEC,
);

Entrambi i metodi costruiscono un frame LOG con i corretti request_id e routing_id e lo inviano attraverso il canale di output. Non sono bloccanti.

Sorgente: capdag-rs/src/bifaci/cartridge_runtime.rs (OutputStream::progress, OutputStream::log).

Convenzioni di avanzamento degli handler

Le cartucce ML seguono una disposizione standard dell’avanzamento:

Queste sono convenzioni, non requisiti del protocollo. Le cartucce di contenuto (PDF, testo) usano disposizioni più semplici perché le loro operazioni sono più rapide.

L’avanzamento finale viaggia in END

L’avanzamento finale della richiesta è un metadato terminale del frame END (L3/L5), non un frame LOG. Quando un handler termina con successo, il runtime imprime progress (predefinito 1.0) e un message facoltativo nella meta di END; un handler può sovrascrivere entrambi tramite OutputStream::finish(progress, message) prima di restituire. Il lato ricevente consegna questo valore come evento di avanzamento finale. Poiché viaggia nel frame terminale, non può entrare in competizione con esso.

Conseguenze:

• Non emettere un LOG finale di avanzamento al 100%. END è l’unico evento di avanzamento terminale autorevole, quindi il completamento non può entrare in competizione con un frame LOG separato.
• END/ERR è l’ULTIMO frame di flusso di una richiesta (L4). Il thread di scrittura della cartuccia lo impone: un frame di flusso che gli arriva dopo che il terminale del flusso è stato scritto (un ProgressSender distaccato, un ticker di keepalive sopravvissuto all’handler) viene scartato e conteggiato (post_terminal), mai scritto. La stessa ragione copre il lato RICEVENTE della legge: un frame che ha incrociato il terminale della sua richiesta sul canale (una concessione di credito, un LOG ritardatario) viene classificato post_terminal da ogni runtime di instradamento (vedi Controllo di flusso §”La corsa alla chiusura”), mantenendo no_route un vero allarme di anomalia di instradamento. Gli scarti sono osservabili nelle statistiche di protocollo del runtime, quindi nulla è silenzioso.
• Un END di successo senza avanzamento terminale esplicito vale 1.0. Un END fallito non sintetizza mai un valore di completamento.

Frame di keepalive

I frame di keepalive risolvono un problema specifico: le chiamate FFI bloccanti (caricamenti di modelli) possono affamare lo scrittore di frame, inducendo l’engine a pensare che la cartuccia sia morta.

Il problema

Il task di scrittura dei frame della cartuccia gira su un thread worker di tokio. Svuota il canale di output e scrive i frame su stdout. Quando un handler chiama una funzione FFI bloccante (ad esempio llama_model_load(), che per i modelli grandi può richiedere minuti), la chiamata gira sullo stesso pool di thread worker di tokio.

Se tutti i worker di tokio sono occupati da chiamate bloccanti, il task di scrittura non può girare. I frame accodati nel canale di output restano lì, anche quelli emessi prima che la chiamata bloccante iniziasse. L’engine non vede arrivare frame su stdout e, dopo 120 secondi di silenzio, fa scattare il suo timeout di attività e termina il task.

La sequenza:

sequenceDiagram
    participant H as Handler
    participant CH as mpsc Channel
    participant WT as Writer Task
    participant E as Engine

    H->>CH: progress(0.25, "Loading model...")
    Note over H: llama_model_load() blocca un worker tokio
    Note over WT: Serve un worker tokio — nessuno disponibile
    Note over CH: Frame bloccati nel canale
    Note over E: 30s... 60s... 90s...
    E->>E: 120s: timeout di attività

run_with_keepalive (Rust)

OutputStream::run_with_keepalive() risolve il problema eseguendo il lavoro bloccante su un pool di thread separato:

pub async fn run_with_keepalive<T: Send + 'static>(
    &self,
    progress: f32,
    message: &str,
    f: impl FnOnce() -> T + Send + 'static,
) -> T

Che cosa fa:

1. Avvia la closure su tokio::task::spawn_blocking, un pool di thread dedicato al lavoro bloccante, separato dai worker asincroni di tokio. La closure gira senza bloccare alcun worker di tokio.
2. Esegue un ciclo di keepalive usando tokio::select! con un tokio::time::interval(30s). Ogni 30 secondi emette un frame di avanzamento con il valore e il messaggio forniti.
3. Restituisce il risultato della closure quando spawn_blocking termina. Il ciclo di keepalive si ferma.

Poiché spawn_blocking libera i worker di tokio, il task di scrittura può girare e scaricare i frame di keepalive su stdout. L’engine vede frame ogni 30 secondi e azzera il suo timeout di 120 secondi.

// Uso in un handler
let model = req.output().run_with_keepalive(
    0.25,
    "Loading model...",
    move || LlamaModel::load_from_file(&model_path, &params),
).await;

Che cosa deve passarci: ogni chiamata bloccante, non solo il caricamento del modello. Un forward pass (una finestra di Whisper, un batch di embedding, una didascalia o una codifica CLIP) è lo stesso lavoro bloccante del caricamento del modello e, eseguito inline su un worker di tokio, ha lo stesso effetto: il ciclo stesso del runtime non viene schedulato, la sonda di heartbeat dell’host resta senza risposta e l’host termina la cartuccia come non reattiva (CARTRIDGE_UNHEALTHY). Sposta il lavoro e i suoi input nella closure (un modello riutilizzato tra le chiamate entra e torna fuori insieme al risultato) ed emetti l’avanzamento dal suo interno attraverso un ProgressSender. Per uno stream di chiamate per chunk su un unico modello caricato, un singolo worker spawn_blocking alimentato tramite un canale segue la stessa regola.

Sorgente: cartridge_runtime.rs (run_with_keepalive).

runWithKeepalive (Swift)

L’equivalente Swift in capdag-objc/Sources/Bifaci/CartridgeRuntime.swift usa la concorrenza strutturata:

1. Avvia un Task in background che emette frame di avanzamento ogni 30 secondi.
2. Attende l’operazione asincrona.
3. Annulla il Task di keepalive quando l’operazione termina.

Il meccanismo è diverso (Swift usa l’annullamento di Task invece di tokio::select!) ma l’effetto è identico: i frame di keepalive fluiscono verso l’engine mentre il lavoro bloccante è in esecuzione.

ProgressSender

ProgressSender è un handle distaccabile per emettere frame di avanzamento e di log dall’interno di una closure spawn_blocking. È Clone + Send + Sync + 'static, quindi può attraversare liberamente i confini tra thread.

#[derive(Clone)]
pub struct ProgressSender {
    sender: Arc<dyn FrameSender>,
    request_id: MessageId,
    routing_id: Option<MessageId>,
}

impl ProgressSender {
    pub fn progress(&self, progress: f32, message: &str);
    pub fn log(
        &self,
        level: &str,
        attribution_class: AttributionClass,
        message: &str,
    );
    pub fn log_for_argument(
        &self,
        level: &str,
        attribution_class: AttributionClass,
        message: &str,
        arg_urn: &str,
    );
}

Creane uno tramite output.progress_sender():

let ps = req.output().progress_sender();
let result = tokio::task::spawn_blocking(move || {
    // Gira sul pool di thread bloccanti
    ps.progress(0.4, "Processing token 50 of 200");
    do_inference(&model, &input, |token_idx, total| {
        ps.progress(0.35 + 0.6 * (token_idx as f32 / total as f32), "Generating...");
    })
}).await.unwrap();

flowchart TD
    Q{"Il lavoro bloccante richiede<br/>un avanzamento granulare?"}
    Q -->|"No: basta tenerlo in vita"| RWK["run_with_keepalive()<br/>Valore fisso ogni 30s"]
    Q -->|"Sì: per token, per passo"| PS["ProgressSender<br/>Emissione dall'interno della closure"]

    RWK --> SB1["spawn_blocking(closure)"]
    PS --> SB2["spawn_blocking(move || {<br/>  ps.progress(...)  <br/>})"]

La differenza fondamentale rispetto a run_with_keepalive: ProgressSender dà alla closure il controllo su quando e quale avanzamento emettere, invece di emettere un valore di keepalive fisso a intervalli fissi. Usa run_with_keepalive per le chiamate bloccanti semplici (caricamenti di modelli). Usa ProgressSender quando il lavoro bloccante stesso può riportare un avanzamento granulare (inferenza per token, elaborazione di più file).

Una pipeline il cui engine prende in prestito dal modello caricato (ad esempio un engine di visione) non può essere restituita da spawn_blocking. L’intera pipeline (caricamento + inferenza) deve allora girare in un’unica closure bloccante, e l’avanzamento per token viene emesso dal suo interno attraverso un ProgressSender.

Sorgente: cartridge_runtime.rs (ProgressSender, progress_sender).

Timeout di attività

L’engine impone un timeout di attività di 120 secondi (DEFAULT_ACTIVITY_TIMEOUT_SECS in capdag-rs/src/orchestrator/executor.rs). Qualsiasi frame sul canale di risposta, comprese le risposte LOG, di avanzamento e di heartbeat, azzera il timer. Se per 120 secondi non arriva alcun frame, l’engine termina il task con ExecutionError::ActivityTimeout.

L’intervallo di keepalive di 30 secondi fornisce un margine di sicurezza di 4x: anche se un frame di keepalive viene ritardato, il successivo arriva ben prima del timeout.

Le singole cap possono sovrascrivere il timeout predefinito tramite il campo activity_timeout nella loro definizione. Questo si usa per le cap che notoriamente hanno lunghi periodi di silenzio che non possono essere interrotti con frame di keepalive.

Mappatura dell’avanzamento lungo la catena di esecuzione

I valori di avanzamento attraversano diversi livelli tra la cartuccia e la UI:

graph TD
    PH["Handler della cartuccia<br/>output.progress(0.5, ...)"] --> LOG["Frame LOG su stdout"]
    LOG --> PHR["CartridgeHostRuntime<br/>(passa invariato)"]
    PHR --> RS["RelaySlave → RelayMaster<br/>(passa invariato)"]
    RS --> EF["execute_fanin<br/>→ callback CapProgressFn"]
    EF --> CI["cap_interpreter<br/>→ mappa sull'intervallo del passo"]
    CI --> DB["Database SQLite"]
    DB --> UI["UI"]

1. Handler della cartuccia: emette l’avanzamento grezzo in [0.0, 1.0].
2. Inoltro delle chiamate peer: se l’handler sta inoltrando l’avanzamento di un peer, mappa il [0.0, 1.0] del peer su un sotto-intervallo (ad esempio [0.0, 0.25]) usando map_progress(value, base, weight).
3. execute_fanin: il ciclo di esecuzione dell’orchestratore legge i frame LOG dal canale di risposta e chiama la callback CapProgressFn con il valore di avanzamento della cartuccia.
4. cap_interpreter: mappa l’avanzamento [0.0, 1.0] della cap sull’intervallo assegnato al passo all’interno dell’avanzamento complessivo del task.
5. Database: il valore mappato viene scritto in SQLite. La UI lo legge.

Vedi Mappatura dell’avanzamento per come ProgressMapper suddivide in modo deterministico gli intervalli di avanzamento tra i passi.

Restrizioni sul logging

Le cartucce non hanno un subscriber di tracing. Quanto segue non funziona all’interno dei processi delle cartucce:

• tracing::info!(), tracing::debug!(), ecc.: non è installato alcun subscriber. I messaggi vengono scartati in silenzio.
• eprintln!(), fputs(stderr): nella sandbox di un’app GUI, stderr finisce in /dev/null. Su macOS, una write() su uno stderr morto può bloccarsi per sempre. Una callback di log di una libreria nativa che chiama fputs(text, stderr) (ad esempio la callback di log predefinita del caricatore di modelli CLIP) può quindi bloccarsi indefinitamente.

L’unico modo per emettere output osservabile da una cartuccia è attraverso i frame LOG tramite output.log(), output.log_for_argument() e output.progress(). Questi inviano frame attraverso il canale stdout, che l’host riceve ed elabora.

Per le librerie FFI che hanno le proprie callback di log (llama.cpp, CLIP/MTMD), la callback deve essere reindirizzata a una funzione che non fa nulla. LlamaBackend::void_logs() sopprime i log di llama; mtmd_log_set() sopprime i log di CLIP/MTMD. Entrambe devono essere chiamate prima di caricare i modelli.