Flujos de trabajo

Los flujos de trabajo dividen las tareas de larga duración en pasos concretos. La salida de cada paso se guarda en un estado compartido en la base de datos. Si PHP se detiene a mitad de un paso, el trabajador lo reintenta con todo el estado anterior intacto.

¿Qué son los flujos de trabajo?

Un flujo de trabajo es una secuencia ordenada de pasos. Cada paso es una clase PHP que recibe el estado acumulado de los pasos anteriores y devuelve datos nuevos para combinarlos con él. El flujo avanza un paso a la vez, y cada transición se ejecuta dentro de una transacción atómica MySQL.

use Queuety\Queuety;

$workflow_id = Queuety::workflow( 'generate_report' )
    ->then( FetchDataHandler::class )
    ->then( CallLLMHandler::class )
    ->then( FormatOutputHandler::class )
    ->dispatch( [ 'user_id' => 42 ] );

Acumulación de estado

La carga inicial que pasas a dispatch() se convierte en el estado inicial. Las claves devueltas por cada paso se combinan con el estado. Las claves que empiezan por _ están reservadas para valores de control como _next_step y no se combinan con el estado público:

dispatch(['user_id' => 42])

Step 0: FetchDataHandler   -> returns ['user_name' => 'Alice']
  state: {user_id: 42, user_name: 'Alice'}

Step 1: CallLLMHandler     -> returns ['llm_response' => '...']
  state: {user_id: 42, user_name: 'Alice', llm_response: '...'}

Step 2: FormatOutputHandler -> returns ['report_url' => '/reports/42.pdf']
  state: {user_id: 42, user_name: 'Alice', llm_response: '...', report_url: '/reports/42.pdf'}

Cada paso recibe el estado acumulado completo. Así es fácil pasar datos entre pasos sin almacenamiento externo.

Reanudación

Los flujos de trabajo son duraderos porque el estado se guarda en la base de datos después de cada paso. Si el trabajador falla durante el paso 1, el flujo se reinicia desde el paso 1 con la salida del paso 0 ya guardada en el estado.

El límite entre pasos es una sola transacción MySQL que, de forma atómica:

1. Actualiza el estado del flujo de trabajo con la salida del paso
2. Marca como completado el trabajo del paso actual
3. Pone en cola el trabajo del siguiente paso

Esto garantiza que nunca se pierda el estado, incluso ante fallos, tiempos de espera agotados o errores fatales PHP.

Ciclo de vida del flujo de trabajo

dispatch() -> running -> step 0 -> step 1 -> ... -> step N -> completed
                  |                   |
                  |                   +-> failed -> retry -> running (from failed step)
                  |
                  +-> paused -> resumed -> running
                  |
                  +-> waiting_for_signal -> signal received -> running
                  |
                  +-> waiting_for_workflows -> dependency settles -> running
                  |
                  +-> cancelled (via cancel_workflow)

Administración de flujos de trabajo

Consultar el estado:

$state = Queuety::workflow_status( $workflow_id );
echo $state->status->value;   // 'running'
echo $state->current_step;    // 1
echo $state->total_steps;     // 3
print_r( $state->waiting_for ); // ['42'] while waiting on another workflow
print_r( $state->wait_details ); // matched/remaining blockers, result key, correlation info

Pausar y reanudar:

Queuety::pause_workflow( $workflow_id );
Queuety::resume_workflow( $workflow_id );

Reintentar un flujo de trabajo fallido desde el paso que falló:

Queuety::retry_workflow( $workflow_id );

Cancelar un flujo de trabajo y ejecutar los controladores de limpieza:

Queuety::cancel_workflow( $workflow_id );

Consulta Cancelación para obtener detalles sobre los controladores de limpieza y lo que ocurre al cancelar un flujo de trabajo.

Opciones del flujo de trabajo

Define la cola, la prioridad y el comportamiento de reintento para todos los pasos de un flujo de trabajo:

use Queuety\Enums\Priority;

Queuety::workflow( 'important_pipeline' )
    ->on_queue( 'pipelines' )
    ->with_priority( Priority::High )
    ->max_attempts( 5 )
    ->on_cancel( PipelineCleanupHandler::class )
    ->prune_state_after( 3 )
    ->then( StepA::class )
    ->then( StepB::class )
    ->dispatch( $payload );

El método on_cancel() registra un controlador de limpieza que se ejecuta cuando se cancela el flujo de trabajo. Consulta Cancelación.

El método prune_state_after() activa la eliminación automática de las salidas antiguas de los pasos para mantener acotado el estado del flujo de trabajo. Consulta Depuración del estado.

Usa compensate_with() para asignar un controlador de compensación a cada paso. Usa compensate_on_failure() para ejecutar automáticamente las compensaciones de los pasos completados tras un fallo del flujo de trabajo, no solo tras su cancelación explícita. Consulta Compensación.

Usa las protecciones para gestionar con seguridad la orquestación de larga duración: version() asigna una versión de la aplicación a la definición, Queuety también almacena un hash determinista de la definición para cada ejecución, idempotency_key() mantiene la durabilidad del envío ante reintentos o solicitudes duplicadas, y max_transitions(), max_for_each_items() y max_state_bytes() hacen fallar las ejecuciones descontroladas antes de que se disparen. Consulta Protecciones.

Tipos de pasos

Queuety admite once tipos de pasos en un flujo de trabajo:

• Cadenas secuenciales con ->then() para ejecutarlas de una en una
• Grupos paralelos con ->parallel() para ejecutarlos de forma concurrente
• Bucle dinámico para cada elemento con ->for_each() para tareas de ramificación descubiertas durante la ejecución y semántica de finalización explícita
• Transferencias asíncronas con ->start_workflows() o ->start_agents() para enviar flujos de trabajo secundarios independientes a partir de elementos descubiertos durante la ejecución
• Ramificación condicional con pasos con nombre y _next_step
• Repeticiones con ->repeat_until() y ->repeat_while() para volver de forma duradera a pasos anteriores con nombre
• Ejecución de flujos de trabajo con ->run_workflow() para la composición anidada
• Demoras duraderas con ->delay() para las demoras basadas en el tiempo
• Señales con ->wait_for_signal(), ->wait_for_signals(), ->wait_for_approval(), ->wait_for_input() y ->wait_for_decision() para activar procesos mediante eventos externos
• Dependencias entre flujos de trabajo con ->wait_for_workflow(), ->wait_for_workflows() y ->wait_for_agents() para la coordinación entre flujos
• Pasos de transmisión con la interfaz StreamingStep para generar fragmentos de salida duraderos

Patrones comunes

Si estás creando sistemas autónomos o semiautónomos, empieza por Orquestación de agentes. Explica cómo se combinan for_each(), start_agents(), wait_for_agents(), wait_for_workflow() y wait_for_decision() en flujos reales de planificación y ejecución, y de revisión humana.