Exécutions

En démarrer une

Une exécution démarre avec un identifiant de définition et une entrée. Le daemon valide l’entrée selon le schéma de la définition quand celle-ci en a un, épingle la version stockée la plus haute, crée l’exécution et sa trace, et répond avec l’exécution telle qu’elle est stockée. À partir de là, l’exécution avance seule : le daemon propose ses étapes aux workers, déroule ses sleeps et ses attentes, et écrit dans le journal ce qui se produit.

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

Envoyez un idempotency_key depuis tout appelant susceptible d’être réessayé. Il est unique par projet tant que l’exécution est conservée, et une nouvelle tentative du même appel renvoie la même exécution au lieu d’en démarrer une seconde. Un webhook qui part deux fois, ou un ordonnanceur de tâches qui redémarre, ne coûte alors rien.

Le journal

Chaque changement d’une exécution est un événement doté d’un numéro de séquence, à partir de zéro et sans trou : l’exécution a démarré, le curseur a avancé, une étape a reçu un bail, a démarré, s’est terminée ou a échoué, l’exécution a attendu puis repris, l’exécution s’est arrêtée. Chacun porte un payload dont la forme dépend de son type. Le journal est l’histoire de l’exécution, et c’est lui que l’interface dessine.

Lisez-le par pages à partir de after, ou gardez l’endpoint des événements ouvert avec Accept: text/event-stream et recevez chaque événement au moment où il est validé. Un client qui s’arrête puis reprend transmet le dernier numéro de séquence qu’il a vu et ne rate rien. Le watch de chaque SDK fait cela pour vous et s’achève avec l’exécution, tandis que wait suit le journal jusqu’au bout et rend la sortie, ou une erreur typée portant l’exécution quand celle-ci n’a pas réussi.

Événements et attentes

Une étape wait_event retient l’exécution jusqu’à l’arrivée d’un événement portant son nom et sa clé évaluée. L’événement entre par une seule porte :

La réponse est l’une des trois issues suivantes. delivered : une exécution attendait et a pris l’événement dans la même transaction, si bien qu’au moment où vous lisez l’accusé de réception, l’exécution a déjà avancé. parked : aucune exécution n’attendait, et l’événement est gardé pour une attente qui s’ouvrira plus tard et déclarera accepter les événements mis de côté. duplicate : l’event_id avait déjà été vu et rien n’a été écrit. Envoyez votre propre identifiant depuis tout ce qui peut partir deux fois.

Une attente qui atteint son échéance fait ce que dit son on_timeout : elle échoue, et son on_error tranche alors, ou elle continue sans sortie. Pour viser une exécution directement, par sa propre attente ouverte, il y a POST /v1/api/runs/{run}/signal ; il prend le même corps sans clé, parce que c’est l’exécution qui fait l’adresse.

Annuler

POST /v1/api/runs/{run}/cancel met fin à une exécution et à toutes les exécutions filles sous elle. Une étape en cours n’est pas interrompue sur le réseau : le worker apprend à son heartbeat suivant que l’exécution est terminée, son contexte le signale au corps, et tout ce que le corps poste ensuite est refusé avec run_terminal. Annuler une exécution déjà terminée ne change rien et renvoie l’exécution telle quelle.

Timers et limites

Les sleeps, les échéances d’attente, les backoffs entre tentatives et les expirations de bail sont autant de timers que le daemon tient. Ils se déclenchent sur l’horloge du daemon, une fois par seconde : un retry_after_ms de cent millisecondes est donc honoré comme un minimum et non comme une valeur exacte. Un redémarrage du daemon n’en perd aucun ; ce sont des lignes, et l’exécution reprend depuis la ligne. Aucune durée ne peut dépasser 3650 jours, et un retry_after_ms plus long envoyé par un worker est refusé avec un 400. Un sleep_until postérieur à l’année 2262 est traité comme une échéance qui n’arrive jamais.

Les max_steps, run_timeout_ms et max_cost_usd d’une définition mettent fin à l’exécution au plafond, en nommant le plafond comme erreur. Le coût est le total de la consommation déclarée par les étapes, tarifée avec la même grille que les spans, de sorte qu’une étape d’agent qui déclare ses tokens est imputée au budget de l’exécution dès qu’elle se termine.

L’exécution dans sa trace

Toute exécution est aussi une trace. Le daemon ouvre un span racine pour l’exécution et un span par tentative, avec l’entrée et la sortie de l’étape, et un corps qui enregistre ses propres spans par le contexte d’étape les place en dessous. L’appel de modèle à l’intérieur d’une étape d’agent se trouve donc sous cette étape, sous cette exécution, et la recherche, la comptabilité des coûts et les scorers qui travaillent sur n’importe quelle trace travaillent sans changement sur la trace d’une exécution. Ces spans sont écrits par le même chemin d’ingestion que les spans d’un SDK : ils sont donc échantillonnés, masqués, tarifés et transférés de la même façon. Les tokens vivent sur les spans de modèle qui les ont dépensés ; le span d’étape porte la somme tarifée dans ses attributs, si bien qu’un total de trace compte chaque token une seule fois. L’exécution porte son trace_id, et l’interface relie les deux.

Le daemon écrit ces spans par une outbox, sur sa propre horloge, après la validation de la ligne qui les a produits. Une trace lue juste après la fin d’une étape peut ne pas encore porter le span de cette étape ; le détail de l’exécution, lui, le porte toujours.

Relire les exécutions

GET /v1/api/runs liste les exécutions d’un projet, de la plus récemment modifiée à la plus ancienne, filtrées par statut ou par définition. GET /v1/api/runs/{run} donne l’exécution et tout ce qu’elle a de durable : chaque tentative avec le détenteur de son bail, ses horodatages, son erreur et sa consommation ; les timers encore en attente ; les attentes encore ouvertes. L’écran Workflows de l’interface est cet endpoint dessiné, avec le journal en direct à côté.

Les deux rendent ce que l’exécution a reçu : les deux sont donc enregistrés dans le journal d’audit, sous list_runs et get_run, de même que la lecture du journal (list_run_events, une fois par requête, y compris pour un flux), le parcours des causes (get_step_cause) et l’annulation (cancel_run). Une lecture refusée est enregistrée au compte du projet qui possède l’exécution. Les appels de worker ne sont pas enregistrés ; voir Workers.

Stockage

Les exécutions vivent dans leur propre fichier SQLite, à côté du stockage des traces : <data>.runs.<ext>, sauf indication contraire de --runs-data. --runs-data off coupe entièrement la surface des exécutions, et chaque endpoint d’exécution répond alors 503. Sauvegardez ce fichier avec le stockage des traces ; mcpd doctor le note comme un groupe à part.

Le stockage des exécutions garde les payloads tels qu’ils ont été envoyés. L’entrée, l’état, la sortie et la corrélation d’une exécution, l’entrée, la sortie et l’erreur de chaque étape, le payload de chaque événement, et les payloads de spans qui attendent encore d’être écrits dans le stockage des traces sont stockés sans masquage, et le user_id de la corrélation d’une exécution est stocké sans hachage. C’est délibéré : ce sont les données de travail du workflow, une étape ultérieure les relit, et une valeur que le masquage aurait réécrite changerait ce que fait l’exécution. Les spans que le daemon écrit dans le stockage des traces passent par le même masquage et le même hachage des identifiants qu’un span envoyé par un SDK : le stockage des traces détient donc la copie masquée. Considérez que le fichier des exécutions contient tout ce que vos étapes ont reçu, et protégez-le et sauvegardez-le en conséquence.

La conservation atteint le stockage des exécutions. La passe horaire qui élague le stockage des traces supprime aussi, sur la même fenêtre par projet, chaque exécution dont la racine s’est terminée avant la coupure, avec ses exécutions filles et tout ce qui est stocké pour elles : tentatives, événements, échéances, attentes, spans en file et événements que ces exécutions ont consommés. Une exécution qui n’est pas terminée n’est jamais supprimée, quel que soit son âge. Les définitions sont gardées.

L’effacement l’atteint aussi. DELETE /v1/api/subjects/{user_id} supprime chaque exécution terminée du projet dont la corrélation porte ce user_id, tel que vous l’avez donné au démarrage de l’exécution, avec tout son arbre d’exécutions filles et tout ce qui est stocké pour elles, et rapporte le compte dans runs_erased. Une exécution de cette personne qui n’est pas terminée est laissée entière, comptée dans runs_in_flight et nommée dans le champ note de la réponse : annulez-la, ou laissez-la se terminer, puis renvoyez l’effacement. Un événement mis de côté pour une exécution qui n’a pas encore atteint son attente est rattaché à sa clé de corrélation, et non à une personne, donc l’effacement ne l’atteint pas. Il est supprimé à son expiration, au bout de 24 heures. Voir Masquage et conservation pour le reste de la réponse.

Ensuite

• Ce qu’une définition peut dire : Workflows.
• Le processus qui sert les étapes : Workers.
• Tous les endpoints d’exécution, générés depuis la spécification : API HTTP, Runs.