Eventos, streaming y reconexión

«Ejecuta las pruebas» es un comando. «Las pruebas terminaron con código 0 sobre el artefacto a-29» es un evento. La diferencia importa: reproducir un evento debe reconstruir una vista, no volver a ejecutar la orden.

Un log durable puede registrar cambios de estado y resultados relevantes. No necesitas guardar cada token del modelo como evento de negocio. Los deltas que aparecen en pantalla son provisionales hasta que se completa y acepta la respuesta.

Cada evento necesita identidad, ámbito, orden y tipo. El payload depende del hecho:

{
  "id": "r-17:42",
  "run_id": "r-17",
  "seq": 42,
  "type": "tool.completed",
  "payload": {"call_id": "c-8", "operation_id": "op-8", "artifact_id": "a-29"}
}

La secuencia ordena eventos de ese run. La identidad permite deduplicar. call_id conecta el resultado con la propuesta, y artifact_id evita incrustar un log enorme. Registra el evento en la misma transacción que el cambio al que se refiere, o utiliza un mecanismo de publicación coherente como la outbox.

Server-Sent Events mantiene una conexión HTTP por la que el servidor envía texto en un formato definido. El navegador dispone de EventSource. Es adecuado para progreso del servidor al cliente; las acciones del usuario pueden seguir usando peticiones HTTP normales.

Un mensaje SSE tiene campos y termina en una línea en blanco:

id: r-17:42
event: tool.completed
data: {"call_id":"c-8","operation_id":"op-8"}

El servidor responde con Content-Type: text/event-stream. El campo id permite al navegador enviar Last-Event-ID al reconectar. SSE no almacena nada por sí mismo: tu servidor debe poder consultar eventos posteriores a ese cursor.

Un serializador sencillo evita que el JSON multilínea rompa el formato:

import json


def encode_sse(event):
    payload = json.dumps(event["payload"], ensure_ascii=False)
    return (f"id: {event['id']}\n"
            f"event: {event['type']}\n"
            f"data: {payload}\n\n")

id y type deben ser valores internos validados, sin saltos de línea. No interpoles texto arbitrario del usuario en cabeceras SSE. El servidor HTTP concreto debe gestionar desconexión, flush y timeouts; este fragmento solo implementa el formato.

Al conectar, autentica al usuario y comprueba que puede ver el run. Interpreta el cursor dentro de ese ámbito, consulta seq > last_seq ORDER BY seq y entrega el historial faltante. Después continúa con eventos nuevos.

Hay una carrera entre consultar y suscribirse: puede llegar un evento en medio. Una solución sencilla es seguir sondeando la base por secuencia; otra combina suscripción con una segunda consulta para cerrar el hueco. Un aviso perdido no debe impedir descubrir eventos durables.

El cliente deduplica porque un evento puede llegar dos veces. También debe detectar huecos o cursors demasiado antiguos. Si la retención ya eliminó el tramo necesario, devuelve un snapshot actual con su secuencia y reinicia la suscripción desde ahí. No ignores silenciosamente eventos perdidos.

Una respuesta del modelo puede llegar en fragmentos. Guarda o muestra esos fragmentos bajo una identidad de intento, pero no la presentes como una respuesta completada mientras falte la señal final y la validación. Si la conexión con el proveedor se corta, el texto parcial puede terminar en mitad de un argumento JSON.

No ejecutes una herramienta con argumentos incompletos. Acumula la llamada hasta que el protocolo indique que está completa, valida y entonces acéptala en el almacenamiento durable. Una UI puede mostrar «preparando llamada», pero el efecto todavía no existe.

Si reintentas una inferencia, no concatentes la salida nueva al texto parcial anterior como si fuese una sola respuesta. Son intentos distintos. Puedes reemplazar el borrador visible y conservar el diagnóstico de ambos sin falsificar la trayectoria aceptada.

Una proyección de interfaz puede implementar un reducer: una función que recibe estado y evento y devuelve el siguiente estado. La función no debería llamar herramientas ni servicios.

def reduce_view(view, event):
    next_view = dict(view)
    if event["type"] == "run.status":
        next_view["status"] = event["payload"]["status"]
    elif event["type"] == "tool.completed":
        next_view["last_operation"] = event["payload"]["operation_id"]
    next_view["cursor"] = event["seq"]
    return next_view

Si al abrir una pestaña se vuelve a enviar un correo, has mezclado reconstrucción con comandos. Reproducir cien veces la misma historia debe dejar la misma vista y producir cero efectos externos.

Un cliente lento no debe bloquear el worker. Separa producción durable de entrega por red. Limita buffers; si el cliente queda demasiado atrás, cierra y permite reconectar con cursor.

Un proxy puede almacenar la respuesta y retrasar el streaming. Configura buffering y timeouts según tu servidor, y usa comentarios heartbeat (: ping) si necesitas mantener viva una conexión inactiva. El heartbeat indica que el transporte sigue vivo, no que el agente esté progresando.

EventSource nativo tiene limitaciones para cabeceras personalizadas. En una aplicación del mismo origen puedes usar una sesión con cookies y protección adecuada. Evita tokens duraderos en la URL porque aparecen en logs e historial. Si necesitas otra estrategia de autenticación, elige un cliente de streaming que la soporte y conserva las mismas reglas de replay.

Genera eventos 1–5, desconecta tras el 3 y produce 6–8 sin cliente. Al reconectar desde 3 deben aparecer 4–8 una vez en la vista, aunque el transporte repita alguno. Repite con un usuario distinto: no debe recibir ni los eventos ni información sobre la existencia del run.

• Server-sent events, WHATWG: formato, EventSource, reconexión y Last-Event-ID.
• Observability primer, OpenTelemetry: distinción de señales operativas y correlación.