Human-in-the-Loop sin condiciones de carrera: lockForUpdate, interrupt() y reconciliación transaccional entre planos

"Cualquier bucle que dependa de la intervención de un operador humano es, por definición, asíncrono, impredecible y una fuente habitual de condiciones de carrera. Lejos de ser un detalle cosmético de interfaz, la gobernanza Human-in-the-Loop es un problema clásico de exclusión mutua, persistencia atómica y tolerancia a fallos transitorios en la red."

1. Resumen Ejecutivo: La Realidad de la Concurrencia Humana

La gobernanza Human-in-the-Loop (HITL) suele presentarse en la literatura de IA generativa como una simple llamada interactiva o un botón en la interfaz de usuario: el agente suspende su ejecución, un operador humano aprueba o rechaza la acción, y el flujo continúa. Sin embargo, en un entorno empresarial con concurrencia concurrente, interfaces web desacopladas y microservicios distribuidos, este enfoque genera condiciones de carrera, inconsistencias de estado y fallos operacionales.

El riesgo más crítico surge ante acciones de mutación no idempotentes o destructivas (por ejemplo, publicar un despliegue en producción, ejecutar una migración de base de datos o emitir un pago): si dos operadores hacen clic simultáneamente en "Aprobar", o si un usuario repite la pulsación durante un pico de latencia, un sistema sin exclusión mutua pesimista puede disparar la ejecución en el runtime dos veces de forma paralela. Del mismo modo, si el worker o proceso web que gestiona la aprobación experimenta un fallo anómalo (muerte por OOM, reinicio de contenedor o timeout del proxy inverso), la solicitud puede quedar permanentemente estancada en un estado transicional.

En este artículo desgranamos la arquitectura transaccional implementada entre el Control Plane (jg_dashboard en Laravel 12) y el Execution Plane (jg_agente en FastAPI/LangGraph): serialización atómica mediante lockForUpdate pesimista, desacoplamiento estricto de las llamadas I/O de red fuera de la transacción de base de datos, idempotencia mediante el hook interrupt() de LangGraph, y un proceso reconciliador (Reaper) gobernado por TTL para autoreparar ejecuciones colgadas en estado PROCESSING.

2. Puntos de Fricción: Las Trampas de la Aprobación Asíncrona

Al auditar implementaciones de agentes en entornos corporativos, identificamos tres fallos estructurales recurrentes en la gestión de HITL:

1. Condición de Carrera en la Interfaz (Double-Click):

   // ANTIPATRÓN TÍPICO: Lectura optimista vulnerable a concurrencia
   $approval = ApprovalRequest::find($id);
   if ($approval->status === 'PENDING') {
       // Si dos peticiones HTTP concurrentes llegan en la misma ventana de milisegundos, ambas validan PENDING
       $approval->update(['status' => 'APPROVED']);
       $agentClient->resumeExecution($approval->execution_id); // Se despacha dos veces en paralelo
   }
   

2. Contaminación Transaccional con Llamadas I/O: Mantener transacciones de base de datos abiertas mientras se realiza una llamada HTTP hacia el agente (AgentClientService) retiene conexiones del pool durante segundos. Si el agente experimenta latencia o sobrecarga, agota el pool de conexiones de la base de datos relacional y propaga una degradación en cascada a todo el sistema.

3. Incertidumbre en el Estado PROCESSING: Para mitigar la duplicidad, muchas arquitecturas marcan la solicitud como PROCESSING antes de invocar al agente. No obstante, si el worker de cola o el proceso web sufre un reinicio abrupto (SIGKILL) o la conexión de red expira sin respuesta, la solicitud queda bloqueada en PROCESSING de forma indefinida, requiriendo intervención manual en la base de datos para no congelar el flujo de negocio.

3. Arquitectura y Solución Técnica: Máquina de Estados y Desacoplamiento

La solución adoptada implementa una máquina de estados estricta en dos fases, separando la mutación transaccional local del tráfico de red asíncrono.

Diagrama de Estados y Flujo Transaccional

┌────────────────────────────────────────────────────────────────────────┐
│                          jg_dashboard (Control Plane)                  │
│                                                                        │
│       ┌─────────┐         lockForUpdate()         ┌────────────┐       │
│       │ PENDING │ ──────────────────────────────► │ PROCESSING │       │
│       └────┬────┘   (atómico en DB::transaction)  └──--──┬─────┘       │
│            ▲                                             │             │
│            │ (Reconciliador / Reaper                     │             │
│            │  si TTL > 120s tras fallo)                  │ I/O Seguro  │
│            └─────────────────────────────────────────────┤ (Post-Commit│
│                                                          │  al Agente) │
│                                                          ▼             │
│                                             ┌────────────────────────┐ │
│                                             │  APPROVED  /  REJECTED │ │
│                                             └────────────────────────┘ │
└──────────────────────────────────────────────────────────┼─────────────┘
                                                           │
                                                           │ HTTP POST (resume)
                                                           │ con Idempotencia
                                                           ▼
┌────────────────────────────────────────────────────────────────────────┐
│                        jg_agente (Execution Plane)                     │
│                                                                        │
│  LangGraph Thread ──► interrupt() ──► Checkpoint ──► Resume Atómico    │
│                       (Wait State)    (PostgreSQL)   (State Unfreeze)  │
└────────────────────────────────────────────────────────────────────────┘

Reglas Arquitectónicas Invariables:

1. Bloqueo Pesimista a Nivel de Fila: Toda transición desde PENDING requiere un lockForUpdate() explícito dentro de un bloque DB::transaction(). Si una segunda petición intenta leer la fila, el motor relacional (PostgreSQL/MySQL) la bloquea hasta que la transacción activa se consolida (commit) o revierte (rollback).
2. I/O Fuera de la Transacción: La llamada HTTP externa hacia jg_agente se ejecuta únicamente después de que la base de datos haya confirmado la transacción y la solicitud se encuentre en estado PROCESSING.
3. Reconciliador Autónomo (Stale Processing Reaper): Un comando de consola programado periódicamente audita todas las solicitudes en PROCESSING cuyo timestamp processing_started_at exceda el umbral de tolerancia (TTL por defecto: 120 segundos). Si no hubo confirmación terminal, el Reaper revierte atómicamente el estado a PENDING para permitir la reevaluación por parte del equipo de operaciones.

4. Código Fuente Comentado

Servicio de Aprobación Transaccional en Laravel 12

// jg_dashboard/app/Services/Hitl/HitlApprovalService.php

namespace App\Services\Hitl;

use App\Models\ApprovalRequest;
use App\Services\Agents\AgentHttpClient;
use Illuminate\Support\Facades\DB;
use InvalidArgumentException;

class HitlApprovalService
{
    public function __construct(
        protected AgentHttpClient $agentClient
    ) {}

    public function resolve(int $approvalId, string $decision, int $userId): ApprovalRequest
    {
        // 1. FASE DE MUTACIÓN ATÓMICA CON BLOQUEO PESIMISTA
        $approval = DB::transaction(function () use ($approvalId, $userId) {
            // lockForUpdate serializa de forma determinista cualquier acceso concurrente
            $record = ApprovalRequest::where('id', $approvalId)
                ->lockForUpdate()
                ->firstOrFail();

            if (! $record->isPending()) {
                throw new InvalidArgumentException(
                    "La solicitud {$approvalId} ya no se encuentra en estado PENDING (estado actual: {$record->status})"
                );
            }

            // Marcamos en estado PROCESSING y registramos timestamp para el Reaper
            $record->markProcessing($userId);

            return $record;
        });

        // 2. FASE DE COMUNICACIÓN I/O EXTERNA (Totalmente desacoplada de la transacción de DB)
        try {
            $resumePayload = [
                'approved' => ($decision === ApprovalRequest::STATUS_APPROVED),
                'resolved_by' => $userId,
            ];

            // Invocamos al agente para reanudar el hilo de LangGraph
            $this->agentClient->resumeExecution($approval->execution_id, $resumePayload);

            // 3. FASE TERMINAL: Consolidamos el estado final en base de datos
            $approval->update([
                'status' => $decision,
                'resolved_at' => now(),
            ]);

            return $approval->fresh();

        } catch (\Throwable $e) {
            // En caso de fallo transitorio en la red, revertimos a PENDING para no atrapar el flujo
            $approval->revertToPendingFromStale();
            throw $e;
        }
    }
}

Proceso de Reconciliación con TTL y Scope Eloquent

// jg_dashboard/app/Models/ApprovalRequest.php

public function scopeStaleProcessing(Builder $query, int $ttlSeconds = 120): Builder
{
    return $query->where('status', self::STATUS_PROCESSING)
        ->where('processing_started_at', '<=', now()->subSeconds($ttlSeconds));
}

public function revertToPendingFromStale(): bool
{
    return $this->update([
        'status' => self::STATUS_PENDING,
        'processing_started_at' => null,
    ]);
}

// jg_dashboard/app/Console/Commands/ReconcileHitlApprovalsCommand.php

public function handle(HitlApprovalService $service): int
{
    $ttl = (int) $this->option('ttl');
    $reconciledCount = $service->reconcileStaleProcessingApprovals($ttl);

    if ($reconciledCount > 0) {
        $this->warn("Reconciliadas {$reconciledCount} solicitudes colgadas en PROCESSING.");
    }

    return self::SUCCESS;
}

Pausa e Idempotencia en LangGraph (Execution Plane)

# jg_agente/app/nodes/human_approval.py

from langgraph.types import interrupt
from app.models.state import AgentState, StepResult

async def human_approval_node(state: AgentState) -> dict:
    """Suspende la ejecución mediante interrupt() hasta recibir autorización."""
    
    # Si la decisión ya fue inyectada en el state durante el resume, actuamos de inmediato
    if state.get("approval_granted") is not None:
        return {"approved": state["approval_granted"]}

    # El runtime de LangGraph persiste el thread_id y congela el workflow
    decision = interrupt({
        "question": "¿Autoriza la ejecución del siguiente paso crítico?",
        "tool_name": state.get("current_tool"),
        "arguments": state.get("current_arguments"),
        "requires_rollback": state.get("requires_rollback", False),
    })

    return {
        "approval_granted": decision.get("approved", False),
        "approval_feedback": decision.get("feedback", ""),
    }

5. Consideraciones de Producción, Trade-offs y Límites

1. Deadlocks en Motores Relacionales: Para prevenir interbloqueos cuando múltiples procesos concurrentes actualizan tareas dependientes, los bloqueos pesimistas deben solicitarse siempre sobre una única fila ordenada por clave primaria (id). Adicionalmente, es indispensable configurar un lock_timeout en la sesión de base de datos (por ejemplo, 5 segundos) para abortar limpiamente en caso de contención excesiva.

2. Idempotencia de la Reanudación en LangGraph: Si el agente procesa exitosamente el resume() pero la red se interrumpe antes de que el Control Plane reciba el código HTTP 200, un reintento posterior del operador encontrará el thread fuera del estado interrupt(). El runtime debe verificar el estado del hilo antes de procesar la orden y devolver una respuesta idempotente basada en el checkpoint actual, impidiendo la re-ejecución del nodo.

3. Mínimo Privilegio y Verificación Criptográfica de Intención: Cada solicitud aprobada almacena el identificador del operador, su dirección IP y el hash criptográfico (SHA-256) de los argumentos inspeccionados. Si el modelo altera los argumentos durante la reanudación del contexto, la discordancia de hash aborta inmediatamente la ejecución.

4. El Caso Límite del Worker Caído y los Límites del Reaper con TTL: ¿Qué ocurre exactamente cuando el worker web o de cola colapsa (por OOM, fallo de hardware o reinicio de contenedor) justo después de marcar el estado en PROCESSING pero antes o durante la invocación HTTP?

   • Si la llamada HTTP nunca llegó a salir, el registro queda huérfano en PROCESSING. Aquí el Reaper actúa correctamente: tras expirar el TTL (120 segundos), revierte el estado a PENDING, permitiendo que el operador vuelva a ver la solicitud disponible y la apruebe.
   • El riesgo de falsos positivos en tareas lentas: Si la llamada HTTP sí llegó al agente y la operación downstream requiere más de 120 segundos en completarse, un Reaper basado únicamente en tiempo local podría interpretar erróneamente un proceso lento como un proceso muerto y revertir la solicitud a PENDING mientras el agente sigue trabajando. Si un segundo operador aprueba la solicitud revertida, se induciría una doble ejecución.
   • Mitigación requerida: Para mitigar este escenario en entornos de alta latencia, el Reaper no debe basarse exclusivamente en un timeout pasivo, sino validar activamente el estado del thread_id en el Execution Plane (mediante un endpoint de inspección de hilo o heartbeat) antes de forzar el retorno a PENDING, o implementar distributed leases renovables periódicamente por el worker que mantiene la conexión activa.

6. Reflexión de Arquitectura: Diseñar para el Fallo Humano y de Red

En sistemas distribuidos, asumir que las interacciones humanas y las conexiones HTTP son fiables conduce a fallos silenciosos en producción. Los operadores hacen doble clic, las pestañas se cierran inesperadamente y las redes fallan de forma no determinista.

La solución no consiste en confiar en la buena voluntad de la interfaz, sino en trasladar la responsabilidad a las garantías que los motores de bases de datos y los autómatas con estado llevan décadas resolviendo: aislamiento ACID en la frontera de entrada, llamadas de red desacopladas de las transacciones y reconciliación autónoma con ventanas temporales explícitas.

7. Enlaces y Referencias en el Repositorio

• Servicio de Gobernanza y Bloqueo Pesimista: jg_dashboard/app/Services/Hitl/HitlApprovalService.php
• Modelo Eloquent y Scopes de Reconciliación: jg_dashboard/app/Models/ApprovalRequest.php
• Comando Artisan del Reaper: jg_dashboard/app/Console/Commands/ReconcileHitlApprovalsCommand.php
• Nodo de Interrupción Asíncrona en LangGraph: jg_agente/app/nodes/human_approval.py
• Tests de Concurrencia y Carrera Pesimista: jg_dashboard/tests/Unit/Services/HitlApprovalServiceConcurrencyTest.php