Ejecuciones

Arrancar una

Una ejecución arranca con el id de una definición y una entrada. El daemon valida la entrada contra el esquema de la definición cuando lo tiene, fija la versión almacenada más alta, acuña la ejecución y su traza, y responde con la ejecución tal como quedó almacenada. A partir de ahí la ejecución avanza sola: el daemon ofrece sus pasos a los workers, recorre sus pausas y sus esperas, y escribe en el diario lo que va ocurriendo.

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

Envía un idempotency_key desde cualquier llamador que se pueda reintentar. Es único por proyecto mientras la ejecución se conserva, y un reintento de la misma llamada responde con la misma ejecución en vez de arrancar una segunda. Así, un webhook que se dispara dos veces, o un lanzador de trabajos que se reinicia, no cuesta nada.

El diario

Cada cambio en una ejecución es un evento con un número de secuencia, desde cero y sin huecos: la ejecución arrancó, el cursor avanzó, un paso recibió un lease, empezó, terminó o falló, la ejecución esperó y se reanudó, la ejecución acabó. Cada uno lleva un payload con la forma de su tipo. El diario es la historia de la ejecución, y es lo que dibuja la interfaz.

Léelo por páginas a partir de after, o mantén abierto el endpoint de eventos con Accept: text/event-stream y recibe cada evento en cuanto se confirma. Un cliente que se para y vuelve a arrancar pasa el último número de secuencia que vio y no se pierde nada. El watch de cada SDK hace esto por ti y termina cuando termina la ejecución, y wait sigue el diario hasta el final y devuelve la salida, o un fallo tipado que lleva dentro la ejecución cuando no salió bien.

Eventos y esperas

Un paso wait_event retiene la ejecución hasta que llega un evento con su nombre y su clave renderizada. El evento entra por una sola puerta:

La respuesta es uno de tres resultados. delivered: había una ejecución esperando y tomó el evento en la misma transacción, así que cuando lees el acuse la ejecución ya ha avanzado. parked: no había ninguna ejecución esperando, y el evento se guarda para una espera que se abra más tarde y declare que acepta eventos aparcados. duplicate: el event_id ya se había visto y no se escribió nada. Envía tu propio id desde cualquier cosa que pueda dispararse dos veces.

Una espera que llega a su plazo hace lo que diga su on_timeout: falla, y entonces decide su on_error, o continúa sin salida. Para enviar a una ejecución directamente, por su propia espera abierta, está POST /v1/api/runs/{run}/signal; acepta el mismo cuerpo sin clave, porque la dirección es la ejecución.

Cancelar

POST /v1/api/runs/{run}/cancel termina una ejecución y toda ejecución hija que cuelgue de ella. Un paso en vuelo no se interrumpe sobre el cable: el worker se entera en su siguiente heartbeat de que la ejecución terminó, su contexto avisa al cuerpo, y lo que el cuerpo publique después se rechaza con run_terminal. Cancelar una ejecución que ya había terminado no cambia nada y responde con la ejecución tal como está.

Temporizadores y límites

Las pausas, los plazos de las esperas, los backoffs de reintento y los vencimientos de los leases son todos temporizadores que guarda el daemon. Se disparan con el reloj del daemon, una vez por segundo, así que un retry_after_ms de cien milisegundos se respeta como al menos eso, no como exactamente eso. Un reinicio del daemon no pierde ninguno; son filas, y la ejecución se reanuda desde la fila. Ninguna duración puede pasar de 3650 días, y un retry_after_ms más largo que envíe un worker se rechaza con un 400. Un sleep_until posterior al año 2262 se guarda como un plazo que no vence nunca.

El max_steps, el run_timeout_ms y el max_cost_usd de una definición terminan la ejecución en el tope, con el tope nombrado como el error. El costo es la suma de lo que valió el consumo que informaron los pasos, con el mismo libro de precios que tarifa los spans, así que un paso de agente que informa de sus tokens se carga contra el presupuesto de la ejecución en cuanto termina.

La ejecución dentro de su traza

Cada ejecución es además una traza. El daemon abre un span raíz para la ejecución y un span por intento, con la entrada y la salida del paso, y un cuerpo que registra sus propios spans a través del contexto del paso los cuelga por debajo. Así, la llamada al modelo dentro de un paso de agente queda bajo ese paso y bajo esa ejecución, y la búsqueda, la contabilidad de costos y los evaluadores que trabajan sobre cualquier traza trabajan igual sobre la traza de una ejecución. Esos spans se escriben por la misma ruta de ingesta que siguen los spans de un SDK, así que se muestrean, se ocultan, se tarifan y se reenvían de la misma forma. Los tokens viven en los spans de modelo que los gastaron; el span del paso lleva en sus atributos la suma ya tarifada, así que el total de una traza cuenta cada token una sola vez. La ejecución lleva su trace_id, y la interfaz enlaza las dos.

El daemon escribe esos spans a través de un outbox y con su propio reloj, después de confirmar la fila que los originó. Una traza leída justo después de que termine un paso puede que todavía no lleve el span de ese paso; el detalle de la propia ejecución siempre lo lleva.

Volver a leer las ejecuciones

GET /v1/api/runs lista las ejecuciones de un proyecto, las actualizadas más recientemente primero, filtradas por estado o por definición. GET /v1/api/runs/{run} es la ejecución y todo lo duradero que hay en ella: cada intento con el dueño de su lease, sus tiempos, su error y su consumo; los temporizadores aún pendientes; las esperas aún abiertas. La pantalla Workflows de la interfaz es este endpoint dibujado, con el diario en vivo al lado.

Los dos devuelven lo que recibió la ejecución, así que los dos quedan anotados en el registro de auditoría, como list_runs y get_run, y también el diario (list_run_events, una vez por solicitud, también cuando es un stream), el recorrido de causas (get_step_cause) y una cancelación (cancel_run). Una lectura rechazada se anota contra el proyecto dueño de la ejecución. Las llamadas de worker no se anotan; consulta Workers.

Almacenamiento

Las ejecuciones viven en su propio archivo SQLite, al lado del almacén de trazas, en <data>.runs.<ext> salvo que --runs-data diga otra cosa, y --runs-data off apaga por completo la superficie de ejecuciones, en cuyo caso todos sus endpoints responden 503. Haz su copia de seguridad junto con la del almacén de trazas; mcpd doctor lo califica como un grupo aparte.

El almacén de ejecuciones guarda los payloads tal como se enviaron. La entrada, el estado, la salida y la correlación de una ejecución, la entrada, la salida y el error de cada paso, el payload de cada evento y los payloads de los spans que todavía esperan a escribirse en el almacén de trazas se guardan sin ocultar, y el user_id de la correlación de una ejecución se guarda sin hash. Es deliberado: son el estado de trabajo del workflow, un paso posterior los vuelve a leer, y un valor que el ocultador hubiera reescrito cambiaría lo que hace la ejecución. Los spans que el daemon escribe en el almacén de trazas pasan por la misma ocultación y el mismo hash de identificadores que un span que envía un SDK, así que el almacén de trazas guarda la copia ocultada. Trata el archivo de ejecuciones como un archivo que guarda todo lo que recibieron tus pasos, y protégelo y haz su copia de seguridad en consecuencia.

La retención llega al almacén de ejecuciones. La pasada horaria que recorta el almacén de trazas borra además, con la misma ventana por proyecto, toda ejecución cuya raíz terminó antes del corte, con sus ejecuciones hijas y todo lo guardado para ellas: intentos, eventos, plazos, esperas, spans en cola y los eventos que esas ejecuciones consumieron. Una ejecución que no ha terminado no se borra nunca, por vieja que sea. Las definiciones se conservan.

La supresión también llega a él. DELETE /v1/api/subjects/{user_id} borra toda ejecución terminada del proyecto cuya correlación lleve ese user_id, tal como lo pasaste al arrancar la ejecución, con todo su árbol de ejecuciones hijas y todo lo guardado para ellas, e informa del recuento como runs_erased. Una ejecución de esa persona que no ha terminado se deja entera, se cuenta en runs_in_flight y se nombra en el note de la respuesta: cancélala, o deja que termine, y envía la supresión otra vez. Un evento aparcado para una ejecución que todavía no ha llegado a su espera va asociado a su clave de correlación y no a una persona, así que la supresión no llega a él; se borra cuando caduca, a las 24 horas. Consulta Ocultación y retención para el resto de la respuesta.

A continuación

• Qué puede decir una definición: Workflows.
• El proceso que sirve los pasos: Workers.
• Todos los endpoints de ejecuciones, generados a partir de la especificación: API HTTP, Runs.