Checkpoints de LangGraph en producción: PostgreSQL, allowlist msgpack y política Fail-Closed en degradación

"En la arquitectura de sistemas críticos existe una máxima clara: si el estado de una sesión no está grabado de forma no volátil en almacenamiento persistente, esa sesión no existe en el mundo real. Persistir estados complejos de agentes en memoria volátil liga la vida del flujo al ciclo de vida de un proceso, y serializar objetos arbitrarios con mecanismos no seguros abre vectores de ejecución remota de código. En entornos de producción, la persistencia atómica y el principio Fail-Closed no son negociables: si el registro duradero se interrumpe, el sistema debe pausar las mutaciones antes que comprometer la auditabilidad."

1. Resumen Ejecutivo: La Brecha entre el Prototipo y el Centro de Datos

LangGraph se ha consolidado como un framework de referencia para orquestar agentes mediante grafos dirigidos de estado. No obstante, existe un salto de ingeniería sustancial entre los entornos de experimentación —donde el estado se almacena en memoria efímera mediante MemorySaver— y una plataforma empresarial sometida a reinicios de contenedores, conmutaciones por error y auditorías regulatorias.

Utilizar almacenamiento exclusivamente en memoria en producción implica que cualquier despliegue, escalado horizontal o reinicio por OOM destruye el contexto y los hilos conversacionales activos. Por otra parte, persistir el estado utilizando serializadores basados en pickle abre una superficie crítica de Ejecución Remota de Código (RCE) si los datos serializados sufren manipulación. Asimismo, si la base de datos relacional experimenta una degradación transitoria y el sistema conmuta temporalmente a un modo de memoria volátil, surge una disyuntiva de arquitectura vital: ¿debe el agente continuar ejecutando acciones de alto impacto sin persistencia duradera?

En este artículo desgranamos la arquitectura de persistencia implementada en jg_agente: un checkpointer dual sobre PostgreSQL (DualPostgresSaver) con aislamiento de I/O síncrono mediante asyncio.to_thread, serialización tipada y segura basada en JsonPlusSerializer/msgpack con listas blancas estrictas de clases (STATE_ALLOWED_CLASSES), vistas canónicas de base de datos para trazabilidad forense, y una política estricta Fail-Closed que bloquea automáticamente cualquier acción de riesgo alto o crítico si la persistencia de checkpoints se encuentra degradada.

2. Puntos de Fricción: Memoria Volátil, RCE por Serialización y el Dilema Fail-Open

Al auditar despliegues reales de agentes de IA, se identifican con frecuencia tres compromisos de diseño:

1. La Fragilidad de MemorySaver en Entornos Distribuidos: Mantener el estado del agente en memoria RAM liga la vida de la sesión a un pod o proceso concreto. En clústeres donde los contenedores se reinician o reprograman de forma dinámica, los usuarios pierden sus hilos de conversación y los flujos con aprobación humana quedan en un estado irrecuperable.

2. Riesgos de RCE mediante Deserialización No Segura: El uso indiscriminado de pickle.dumps() para almacenar estados complejos de Python sin restricciones permite que cualquier manipulación en la base de datos o en el canal de transporte derive en la instanciación de objetos maliciosos y ejecución remota de código en el servidor.

3. El Peligro del Enfoque Fail-Open ante Caídas de Base de Datos: Cuando PostgreSQL pierde conectividad, muchos sistemas conmutan silenciosamente a almacenamiento en memoria para "mantener el servicio arriba" (Fail-Open). Sin embargo, permitir que el agente continúe ejecutando transferencias, escrituras en disco o despliegues sin registro persistente elimina la trazabilidad de auditoría y hace inviable la compensación mediante Sagas o HITL.

3. Arquitectura y Solución Técnica: DualPostgresSaver y Conmutación Segura

En jg_agente, la persistencia del estado se desacopla en un motor dual de persistencia con serialización tipada y gobernanza Fail-Closed.

Diagrama de Flujo: Persistencia Dual y Política Fail-Closed

                             ┌────────────────────────┐
                             │  LangGraph State Step  │
                             └───────────┬────────────┘
                                         │
                                         ▼
                             ┌────────────────────────┐
                             │  JsonPlusSerializer    │
                             │  Allowlist Validation  │
                             │ (STATE_ALLOWED_CLASSES)│
                             └───────────┬────────────┘
                                         │
                        ¿Conexión PostgreSQL Operativa?
                                ├─── SÍ ──────────────────────────────────────┐
                                │                                             │
                                ▼ NO (Degradación)                            ▼
                    ┌────────────────────────┐                   ┌─────────────────────────┐
                    │      MemorySaver       │                   │   DualPostgresSaver     │
                    │  (Almacén Volátil)     │                   │  (asyncio.to_thread)    │
                    │ checkpoint_degraded=T  │                   │                         │
                    └───────────┬────────────┘                   │  Tablas y Vistas:       │
                                │                                │  - core_checkpoints     │
                                ▼                                │  - core_checkpoint_blobs│
                    ┌────────────────────────┐                   │  - core_checkpoints_view│
                    │ SecurityGuard.evaluate │                   └────────────┬────────────┘
                    └───────────┬────────────┘                                │
                                │                                             ▼
                 ¿Nivel de Riesgo de la Tool?                         [ESTADO PERSISTIDO]
                 ├── READ_ONLY / LOW   ──► ALLOW                      Trazabilidad Garantizada
                 │
                 └── HIGH / CRITICAL   ──► DENY (Fail-Closed)
                     "Operación bloqueada: Checkpointer degradado"

Pilares de la Solución:

1. Aislamiento de I/O Síncrono del Driver Relacional: Los drivers síncronos de PostgreSQL (psycopg2 o checkpointers síncronos) bloquean el Event Loop si se invocan directamente desde corrutinas asíncronas de FastAPI. DualPostgresSaver traslada las operaciones de guardado y lectura a hilos dedicados mediante asyncio.to_thread(), evitando que la latencia de disco de la base de datos degrade el rendimiento de peticiones concurrentes.

2. Serialización Segura y Tipada: El serializador sustituye pickle por una combinación estricta de JSON y msgpack, validando cada objeto contra una lista blanca explícita de tipos de dominio (AgentState, StepResult, RiskLevel, AIMessage, etc.). Cualquier clase no registrada es rechazada de inmediato.

3. Fail-Closed Ante Modo Degradado: Si PostgreSQL pierde conectividad, el agente activa el fallback en memoria para permitir que tareas básicas de solo lectura o consultas conversacionales informativas continúen respondiendo. No obstante, si el agente intenta invocar una herramienta clasificada como HIGH o CRITICAL, el SecurityGuard bloquea la operación: ninguna mutación con impacto colateral se ejecuta sin persistencia atómica en disco.

4. Código Fuente Comentado

Persistencia Dual y Ejecución en Threadpool Aislado

# jg_agente/app/core/checkpointer.py

import asyncio
from typing import Any, Optional
from langgraph.checkpoint.base import BaseCheckpointSaver, Checkpoint, CheckpointMetadata
from langgraph.checkpoint.memory import MemorySaver

class DualPostgresSaver(BaseCheckpointSaver):
    """Checkpointer de producción sobre PostgreSQL con aislamiento en Threadpool y modo degradado."""

    def __init__(self, sync_saver: BaseCheckpointSaver, fallback_saver: MemorySaver):
        super().__init__()
        self.sync_saver = sync_saver
        self.fallback_saver = fallback_saver
        self.is_degraded = False

    async def aput(
        self,
        config: dict[str, Any],
        checkpoint: Checkpoint,
        metadata: CheckpointMetadata,
        new_versions: dict[str, Any],
    ) -> dict[str, Any]:
        """Persiste el checkpoint de forma asíncrona aislando el I/O en un threadpool."""
        
        if not self.is_degraded:
            try:
                # Trasladamos la llamada bloqueante de psycopg a un hilo de trabajo
                return await asyncio.to_thread(
                    self.sync_saver.put,
                    config,
                    checkpoint,
                    metadata,
                    new_versions,
                )
            except Exception as exc:
                # Conmutación automática a modo degradado
                self.is_degraded = True
                logger.critical(f"Fallo en PostgreSQL checkpointer. Conmutando a MemorySaver degradado: {exc}")

        # Guardado en memoria volátil de fallback
        return await self.fallback_saver.aput(config, checkpoint, metadata, new_versions)

Serializador con Allowlist Estricta contra Ataques RCE

# jg_agente/app/core/checkpointer.py

# Lista blanca estricta de clases autorizadas para deserialización
STATE_ALLOWED_CLASSES = {
    "app.models.state.AgentState",
    "app.models.state.StepResult",
    "app.models.security.RiskLevel",
    "langchain_core.messages.ai.AIMessage",
    "langchain_core.messages.human.HumanMessage",
    "langchain_core.messages.system.SystemMessage",
}

class SafeStateSerializer:
    """Serializador basado en JSON/msgpack que rechaza tipos no autorizados."""

    @classmethod
    def loads(cls, raw_data: bytes) -> Any:
        obj = msgpack.unpackb(raw_data, raw=False)
        cls._verify_allowed_types(obj)
        return obj

    @classmethod
    def _verify_allowed_types(cls, data: Any) -> None:
        if isinstance(data, dict):
            type_name = data.get("__class__")
            if type_name and type_name not in STATE_ALLOWED_CLASSES:
                raise SecurityException(
                    f"Intento de deserializar tipo no autorizado '{type_name}'. Rechazado por seguridad."
                )
            for v in data.values():
                cls._verify_allowed_types(v)

Inyección de Estado Degradado y Bloqueo Fail-Closed en el Grafo

# jg_agente/app/core/graph.py (Inyección automática en el inicio del flujo)

async def initialize_state_node(state: AgentState, checkpointer: DualPostgresSaver) -> dict:
    """Inyecta en el estado si el almacenamiento opera actualmente en modo degradado."""
    return {
        "checkpoint_degraded": checkpointer.is_degraded,
    }

# jg_agente/app/security/guard.py (Bloqueo determinista Fail-Closed)

if checkpoint_degraded and tool.risk_level in [RiskLevel.HIGH, RiskLevel.CRITICAL]:
    return EvaluationResult(
        decision=PolicyDecision.DENY,
        reason=(
            f"Operación '{tool.name}' ({tool.risk_level.value}) bloqueada: "
            "la persistencia de checkpoints está degradada (Fail-Closed Policy)"
        ),
    )

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

1. Vistas Canónicas para Auditoría Forense: El esquema de base de datos define vistas normalizadas (core_checkpoints_view, core_checkpoint_writes_view) que decodifican los payloads binarios y exponen el historial paso a paso para herramientas de observabilidad corporativas y el panel de control de jg_dashboard.

2. Reanudación tras la Recuperación de PostgreSQL: Cuando el clúster relacional se recupera, un healthcheck periódico en segundo plano sondea la conectividad. Una vez restablecida, la bandera is_degraded se restablece a False, permitiendo nuevamente las operaciones críticas. Los checkpoints acumulados en memoria durante el incidente se sincronizan con la base de datos de forma controlada.

3. Trade-off Fundamental: Disponibilidad Degradada vs. Integridad Transaccional: La aplicación estricta de la política Fail-Closed implica asumir una decisión arquitectónica deliberada:

   • En un modelo Fail-Open, se prioriza la disponibilidad a toda costa: el agente continuaría ejecutando mutaciones incluso sin base de datos. La consecuencia es que, ante un reinicio del contenedor, se pierde la traza forense, se quiebra la idempotencia y resulta imposible reconstruir el estado para un rollback de Sagas o una aprobación HITL.
   • En el modelo Fail-Closed de jg_agente, la disponibilidad del sistema se sacrifica de forma parcial y controlada para salvaguardar la consistencia de los datos y el cumplimiento de auditoría. Las tareas informativas (READ_ONLY o LOW) continúan atendiendo a los usuarios, pero cualquier acción con consecuencias externas persistentes queda vetada hasta que se restablezca el almacenamiento duradero.

6. Reflexión de Arquitectura: Sin registro no hay producción

En sistemas transaccionales, la disponibilidad es un objetivo central, pero la consistencia y la auditabilidad son invariantes de diseño. Ante una degradación de almacenamiento, operar a ciegas bajo la justificación de mantener la continuidad del servicio introduce riesgos inasumibles.

En el diseño de agentes de IA, el enfoque Fail-Open no mitigado genera estados corruptos y pérdida de trazabilidad. En jg_agente adoptamos una postura clara: ante la pérdida de persistencia atómica, se emite un DENY controlado antes que ejecutar mutaciones sin respaldo duradero. La verdadera resiliencia consiste en preservar la integridad y la consistencia del sistema en cualquier circunstancia.

7. Enlaces y Referencias en el Repositorio

• Módulo del Checkpointer Dual: jg_agente/app/core/checkpointer.py
• Evaluación Fail-Closed en el Guardián: jg_agente/app/security/guard.py
• Nodo de Evaluación de Riesgo: jg_agente/app/nodes/risk_assess.py
• Definición del Grafo y Estado: jg_agente/app/core/graph.py y jg_agente/app/models/state.py
• Suite de Pruebas de Persistencia y Fail-Closed: jg_agente/tests/test_security_guard.py