Esecuzioni

Avviarne una

Un’esecuzione parte da un id di definizione e da un input. Il daemon convalida l’input rispetto allo schema della definizione, quando ce n’è uno, fissa la versione più alta fra quelle archiviate, crea l’esecuzione e la sua traccia, e risponde con l’esecuzione così com’è archiviata. Da lì in poi l’esecuzione va avanti da sola: il daemon offre i suoi passi ai worker, percorre i suoi sleep e le sue attese, e scrive nel diario quello che succede.

// Start a run from anywhere that holds an execute key, answer the wait from a
// webhook, then follow the journal to the end and read the run.
const run = await mcpd.runs.start({
  definitionId: 'support_ticket',
  input: { ticket: 'Q-42' },
})

await mcpd.events.publish({
  event: 'approved',
  key: 'Q-42',
  payload: { by: 'ops' },
  eventId: 'hook-1', // a repeat of this id is a duplicate, not a second event
})

for await (const event of mcpd.runs.watch(run.runId)) {
  console.log(event.seq, event.kind)
}
const detail = await mcpd.runs.get(run.runId)
console.log(detail.run.status, detail.run.output)

# Start a run from anywhere that holds an execute key, answer the wait from a
# webhook, then follow the journal to the end and read the run.
run = mcpd.runs.start(support, input={"ticket": "Q-42"})

# A repeat of this event id is a duplicate, not a second event.
mcpd.events.publish("approved", "Q-42", payload={"by": "ops"}, event_id="hook-1")

for event in mcpd.runs.watch(run.run_id):
    print(event.seq, event.kind)
detail = mcpd.runs.get(run.run_id)
print(detail.run.status, detail.run.output)

// Start a run from anywhere that holds an execute key, answer the wait from
// a webhook, then follow the journal to the end and read the run.
let start = StartRun::new("support_ticket").input(json!({ "ticket": "Q-42" }));
let run = mcpd.start_run(start).await?;

// A repeat of this event id is a duplicate, not a second event.
let payload = json!({ "by": "ops" });
let approval = PublishEvent::new("approved", "Q-42").payload(payload).event_id("hook-1");
mcpd.publish_event(approval).await?;

mcpd.watch_run(&run.run_id, -1, |e| println!("{} {:?}", e.seq, e.kind)).await?;
let detail = mcpd.get_run(&run.run_id).await?;
println!("{:?} {:?}", detail.run.status, detail.run.output);

// Start a run from anywhere that holds an execute key, answer the wait from
// a webhook, then follow the journal to the end and read the run.
run, err := m.StartRun(ctx, mcpd.StartRun{
	Definition: "support_ticket",
	Input:      map[string]any{"ticket": "Q-42"},
})
if err != nil {
	panic(err)
}
// A repeat of this event id is a duplicate, not a second event.
_, err = m.PublishEvent(ctx, mcpd.PublishEvent{
	Event:   "approved",
	Key:     "Q-42",
	Payload: map[string]any{"by": "ops"},
	EventID: "hook-1",
})
if err != nil {
	panic(err)
}
_ = m.WatchRun(ctx, run.RunID, -1, func(e mcpd.RunEvent) error {
	fmt.Println(e.Seq, e.Kind)
	return nil
})
detail, _ := m.GetRun(ctx, run.RunID)
fmt.Println(detail.Run.Status, string(detail.Run.Output))

// Start a run from anywhere that holds an execute key, answer the wait from
// a webhook, then follow the journal to the end and read the run.
val run = mcpd.startRun("support_ticket", input = mapOf("ticket" to "Q-42"))

// A repeat of this event id is a duplicate, not a second event.
mcpd.publishEvent("approved", "Q-42", payload = mapOf("by" to "ops"), eventId = "hook-1")

mcpd.watchRun(run.runId) { event -> println("${event.seq} ${event.kind}") }
val detail = mcpd.getRun(run.runId)
println("${detail.run.status} ${detail.run.output}")

# Start a run from anywhere that holds an execute key, answer the wait from a
# webhook, then follow the journal to the end and read the run.
{:ok, run} = Mcpd.start_run(mcpd, support, input: %{"ticket" => "Q-42"})

# A repeat of this event id is a duplicate, not a second event.
{:ok, _} =
  Mcpd.publish_event(mcpd, "approved", "Q-42",
    payload: %{"by" => "ops"},
    event_id: "hook-1"
  )

mcpd |> Mcpd.watch_run(run.run_id) |> Enum.each(&IO.inspect({&1.seq, &1.kind}))
{:ok, detail} = Mcpd.get_run(mcpd, run.run_id)
IO.inspect({detail.run.status, detail.run.output})

Manda un idempotency_key da qualunque chiamante che possa riprovare. È unico per progetto finché l’esecuzione è tenuta, e una seconda chiamata uguale risponde con la stessa esecuzione invece di avviarne una seconda. Così un webhook che scatta due volte, o un job runner che riparte, non costa niente.

Il diario

Ogni modifica di un’esecuzione è un evento con un numero di sequenza, da zero e senza buchi: l’esecuzione è partita, il cursore si è mosso, un passo è stato dato in lease, è partito, è finito o è fallito, l’esecuzione ha aspettato e ha ripreso, l’esecuzione è finita. Ognuno porta un payload la cui forma dipende dal suo tipo. Il diario è la storia dell’esecuzione, ed è quello che l’interfaccia disegna.

Leggilo a pagine a partire da after, oppure tieni aperto l’endpoint degli eventi con Accept: text/event-stream e ricevi ogni evento appena viene confermato. Un client che si ferma e riparte passa l’ultimo numero di sequenza che ha visto e non perde niente. Il watch di ogni SDK fa questo per te e finisce quando finisce l’esecuzione, mentre wait segue il diario fino in fondo e ti restituisce l’output, oppure un fallimento tipizzato che porta con sé l’esecuzione quando non è andata a buon fine.

Eventi e attese

Un passo wait_event tiene ferma l’esecuzione finché non arriva un evento con il suo nome e la sua chiave resa dal template. L’evento entra da un solo ingresso:

La risposta è uno di tre esiti. delivered: un’esecuzione stava aspettando e ha preso l’evento nella stessa transazione, quindi quando leggi la ricevuta l’esecuzione si è già mossa. parked: nessuna esecuzione stava aspettando, e l’evento resta da parte per un’attesa che si apre più tardi e dichiara di accettare eventi parcheggiati. duplicate: l’event_id era già stato visto e non è stato scritto niente. Manda un id tuo da qualunque cosa possa scattare due volte.

Un’attesa che arriva alla sua scadenza fa quello che dice il suo on_timeout: fallisce, e allora decide il suo on_error, oppure prosegue senza output. Per mandare un segnale a una sola esecuzione, attraverso la sua attesa aperta, c’è POST /v1/api/runs/{run}/signal; prende lo stesso corpo senza chiave, perché l’indirizzo è l’esecuzione stessa.

Annullare

POST /v1/api/runs/{run}/cancel chiude un’esecuzione e ogni esecuzione figlia sotto di essa. Un passo già in corso non viene interrotto a metà: il worker scopre al suo heartbeat successivo che l’esecuzione è finita, il suo contesto lo segnala al corpo, e qualunque cosa il corpo mandi dopo viene rifiutata con run_terminal. Annullare un’esecuzione già finita non cambia niente e risponde con l’esecuzione così com’è.

Timer e limiti

Gli sleep, le scadenze delle attese, i backoff dei nuovi tentativi e le scadenze dei lease sono tutti timer che tiene il daemon. Scattano sull’orologio del daemon, una volta al secondo, quindi un retry_after_ms di cento millisecondi è onorato come almeno tanto, non come esattamente tanto. Un riavvio del daemon non ne perde nessuno: sono righe, e l’esecuzione riprende dalla riga. Nessuna durata può superare 3650 giorni, e un retry_after_ms più lungo mandato da un worker viene rifiutato con un 400. Uno sleep_until oltre l’anno 2262 viene tenuto come una scadenza che non arriva mai.

I max_steps, run_timeout_ms e max_cost_usd di una definizione chiudono l’esecuzione al tetto, con il tetto stesso nominato come errore. Il costo è la somma dei consumi che i passi riferiscono, valorizzati sullo stesso listino con cui si valorizzano gli span, così un passo di agente che riferisce i suoi token pesa sul budget dell’esecuzione nel momento in cui finisce.

L’esecuzione dentro la sua traccia

Ogni esecuzione è anche una traccia. Il daemon apre uno span radice per l’esecuzione e uno span per tentativo, con l’input e l’output del passo, e un corpo che registra span propri attraverso il contesto del passo li mette lì sotto. Così la chiamata al modello dentro un passo di agente sta sotto quel passo, sotto quell’esecuzione, e la ricerca, la contabilità dei costi e gli scorer che lavorano su una traccia qualsiasi lavorano senza cambiamenti sulla traccia di un’esecuzione. Quegli span vengono scritti attraverso lo stesso percorso di ingest che seguono gli span di un SDK, quindi sono campionati, oscurati, valorizzati e inoltrati allo stesso modo. I token stanno sugli span del modello che li ha spesi; lo span del passo porta nei suoi attributi la somma già valorizzata, così il totale di una traccia conta ogni token una volta sola. L’esecuzione porta il suo trace_id, e l’interfaccia lega le due cose.

Il daemon scrive quegli span attraverso un outbox, sul proprio orologio, dopo che la riga che li ha prodotti è stata confermata. Una traccia letta un istante dopo che un passo è finito può non portare ancora lo span di quel passo; il dettaglio dell’esecuzione lo porta sempre.

Rileggere le esecuzioni

GET /v1/api/runs elenca le esecuzioni di un progetto, a partire da quella aggiornata più di recente, filtrate per stato o per definizione. GET /v1/api/runs/{run} è l’esecuzione e tutto quello che di lei è durevole: ogni tentativo con il proprietario del suo lease, i tempi, l’errore e il consumo; i timer ancora in sospeso; le attese ancora aperte. La schermata Workflow dell’interfaccia è questo endpoint disegnato, con il diario vivo accanto.

Tutti e due restituiscono quello che è stato dato all’esecuzione, quindi tutti e due vengono registrati nel registro di audit, come list_runs e get_run, e così il diario (list_run_events, una volta per richiesta, anche quando è uno stream), la risalita alla causa (get_step_cause) e un annullamento (cancel_run). Una lettura rifiutata viene registrata sul progetto a cui appartiene l’esecuzione. Le chiamate dei worker non vengono registrate; vedi Worker.

Storage

Le esecuzioni vivono in un file SQLite proprio, accanto allo storage delle tracce: <data>.runs.<ext>, a meno che --runs-data non dica altro, e --runs-data off spegne del tutto la superficie delle esecuzioni, nel qual caso ogni endpoint delle esecuzioni risponde 503. Fanne il backup insieme allo storage delle tracce; mcpd doctor lo valuta come gruppo a sé.

Lo storage delle esecuzioni tiene i payload così come sono stati mandati. L’input, lo stato, l’output e la correlazione di un’esecuzione, l’input, l’output e l’errore di ogni passo, il payload di ogni evento, e i payload degli span che aspettano ancora di essere scritti nello storage delle tracce sono archiviati senza oscuramento, e il user_id nella correlazione di un’esecuzione è archiviato senza hash. È voluto: sono lo stato di lavoro del workflow, un passo successivo li rilegge, e un valore che l’oscuratore avesse riscritto cambierebbe quello che fa l’esecuzione. Gli span che il daemon scrive nello storage delle tracce passano per lo stesso oscuramento e lo stesso hash degli identificatori di uno span mandato da un SDK, quindi lo storage delle tracce tiene la copia oscurata. Considera il file delle esecuzioni come un file che tiene tutto quello che è stato dato ai tuoi passi, e proteggilo e fanne il backup di conseguenza.

La conservazione raggiunge lo storage delle esecuzioni. La passata oraria che sfoltisce lo storage delle tracce cancella anche, sulla stessa finestra per progetto, ogni esecuzione la cui radice è finita prima del taglio, con le sue esecuzioni figlie e tutto quello che è archiviato per loro: tentativi, eventi, scadenze, attese, span in coda e gli eventi che quelle esecuzioni hanno consumato. Un’esecuzione che non è finita non viene mai cancellata, per quanto sia vecchia. Le definizioni restano.

Anche la cancellazione lo raggiunge. DELETE /v1/api/subjects/{user_id} cancella ogni esecuzione finita del progetto la cui correlazione porta quel user_id, nella forma con cui hai avviato l’esecuzione, con tutto il suo albero di esecuzioni figlie e tutto quello che è archiviato per loro, e riporta il conteggio come runs_erased. Un’esecuzione di quella persona che non è finita resta intera, contata in runs_in_flight e nominata nella note della risposta: annullala, oppure lasciala finire, e manda di nuovo la cancellazione. Un evento parcheggiato per un’esecuzione che non è ancora arrivata alla sua attesa ha come chiave la sua chiave di correlazione, non una persona, quindi la cancellazione non lo raggiunge; viene cancellato quando scade, dopo 24 ore. Per il resto della risposta, vedi Oscuramento e conservazione.

Poi

• Che cosa può dire una definizione: Workflow.
• Il processo che serve i passi: Worker.
• Ogni endpoint delle esecuzioni, generato dalla specifica: API HTTP, Runs.