Operación: métricas, eventos, inbox y entregas

Esta página cubre dos cosas: las métricas de runtime que expone todo Bot (parte de @botmesh/core) y los paquetes operativos opcionales (journal de eventos, inbox, CRM, campañas...). Para enviar un mensaje a muchos chats o iniciar un flujo desde un job, ve Difusiones e Iniciar un flujo desde código. Para sobrevivir a caídas sin perder ni duplicar mensajes (inbox/outbox durables, reintentos, cola de muertos), ve Procesamiento durable.

Métricas de runtime

bot.metrics() devuelve una foto en memoria de contadores que solo el motor conoce. Es JSON plano y no se persiste nada: los contadores son monótonos desde que se construyó el bot, o desde bot.resetMetrics() (que los pone a cero y conserva el uptime).

const m = bot.metrics()
// {
//   startedAt, uptimeMs, uptime,   // uptime: '3d 4h', '5m 12s'...
//   conversations: { started, completed, ended, guarded, cancelled, timedOut, retriesExceeded, errored, avgDurationMs, active },
//   flows: { registro: { started, completed, ended, guarded, cancelled, timedOut, retriesExceeded, errored, avgDurationMs } },
//   messages: { in, out, outFailed },
//   dropped: { duplicate, blocked, handoff },
//   presenceSkipped: { typing, markRead },
//   triggered,   // llamadas a bot.trigger() que iniciaron una conversación
//   errors,      // errores reportados en bot.on('error')
// }

• messages.in cuenta los mensajes e interacciones de entrada aceptados tras las listas de acceso y el dedupe. out / outFailed cuentan los envíos que el provider aceptó o rechazó.
• dropped.duplicate cuenta las reentregas que descarta el dedupe de entrada, blocked los mensajes, interacciones y llamadas que descartan la blacklist / whitelist (cada entrega, porque el dedupe nunca registra lo bloqueado), y handoff los mensajes que no se enrutan porque el chat está en manos de un humano.
• presenceSkipped.typing / markRead cuentan las señales de presencia (ctx.typing(), ctx.recording(), el indicador automático de escritura, ctx.markRead()) omitidas porque el transporte estaba desconectado o parado (de mejor esfuerzo, ver Canales).
• flows va por nombre de conversación/flujo. avgDurationMs es la duración media de las conversaciones que ya terminaron (0 si ninguna terminó).
• Un flujo que termina su función y un goto cuentan como completed; c.end() cuenta como ended; un .guard() que falla (despedida o redirect) cuenta como guarded; /cancel, /reset, un reemplazo o bot.stop() cuentan como cancelled.

Prometheus

toPrometheus(snapshot, { prefix?, labels? }) renderiza una foto en el formato de texto de Prometheus (0.0.4), sin dependencias. Sírvela con PROMETHEUS_CONTENT_TYPE:

import { createServer } from 'node:http'
import { toPrometheus, PROMETHEUS_CONTENT_TYPE } from '@botmesh/core'

createServer((_req, res) => {
  res.setHeader('Content-Type', PROMETHEUS_CONTENT_TYPE)
  res.end(toPrometheus(bot.metrics(), { labels: { bot: 'ventas' } }))
}).listen(9464)

El prefijo por defecto es botmesh_ (prefix: 'botwa_' conserva los nombres de las versiones alpha). Familias: uptime_seconds, conversations_active, conversations_started_total, conversations_ended_total{outcome} (completed / ended / guarded / cancelled / timed_out / retries_exceeded / error), conversation_duration_avg_seconds, flow_started_total{flow}, flow_ended_total{flow,outcome}, flow_duration_avg_seconds{flow}, messages_in_total, messages_out_total{result} (sent / failed), inbound_dropped_total{reason} (duplicate / blocked / handoff), presence_skipped_total{kind} (typing / mark_read), triggers_total y errors_total. Un prefijo o un nombre de label inválido lanza TypeError.

Eventos de conversación

bot.on('conversation:started', ({ id, name, chatId, userId, key }) => { /* ... */ })
bot.on('conversation:ended', ({ id, name, chatId, outcome, completed, durationMs, goto }) => { /* ... */ })

• conversation:started se emite cada vez que se entra en un flujo o una bot.conversation: un trigger que coincide, bot.trigger(), una reanudación dormida o un c.goto().
• conversation:ended añade outcome ('completed' | 'ended' | 'guarded' | 'cancelled' | 'timedOut' | 'retriesExceeded' | 'error'), completed (atajo de outcome === 'completed'), durationMs y, si terminó con un goto, su destino en goto (le sigue un nuevo started).
• id es único por instancia de conversación, así que empareja cada ended con su started. key es la clave del runtime (chatId, o chatId:userId para un participante de grupo). Cuando una conversación se reemplaza, su ended (outcome: 'cancelled') se emite antes del started de la nueva.
• Un listener que lanza se reporta en bot.on('error'); nunca rompe la conversación.

Paquetes operativos

Paquetes experimentales — ninguno está publicado

@botmesh/events es experimental: una implementación real y con tests, pero todavía privada (no está en npm). Los paquetes de producto (@botmesh/inbox, @botmesh/crm, @botmesh/hub, @botmesh/campaigns y @botmesh/deployments) son reales y persistentes: cada uno tiene un contrato de almacenamiento, una implementación de referencia en RAM y un adaptador Kv* sobre cualquier DurableKv del core (SQLite, PostgreSQL, Redis, Redis Cluster o archivo), con una suite de conformidad compartida y suites opt-in contra servidores reales. Su API todavía puede cambiar mientras la alpha no se publique.

Los paquetes operativos son agnósticos de canal y están aislados por tenant.

Procesamiento de eventos

Usa MemoryEventJournal en tests. FileEventJournal.open('./data/events.json') persiste el journal de una sola instancia con snapshots atómicos, idempotencia, leases, reintentos y replay. Para múltiples réplicas implementa EventJournal sobre PostgreSQL, Redis o un broker.

EventWorker reclama eventos, invoca el handler, confirma el trabajo exitoso y mueve a la cola de mensajes muertos (DLQ) los que agotan sus reintentos.

Cada claim entrega un fencing token (record.fencingToken, estrictamente creciente por evento y nunca reutilizado, ni siquiera tras un replay). El worker lo devuelve en ack/fail y el journal rechaza un token superado con StaleLeaseError: un worker cuyo lease expiró mientras su handler seguía corriendo no puede completar, reintentar ni mandar a la DLQ un evento que otro worker ya reclamó. runBatch() los cuenta como stale ({ processed, failed, stale }) y se salta un registro cuyo lease ya venció antes de empezar su handler; lo devuelve con journal.release(eventId, { now, fencingToken }), así que saltárselo no gasta un intento. Los handlers siguen ejecutándose al menos una vez, así que hazlos idempotentes (para eso está el idempotencyKey del envelope). new EventWorker(journal, { leaseMs, clock }) fija la duración del lease y la fuente de tiempo. Un adaptador de EventJournal sobre PostgreSQL o Redis debe comprobar el token (en ack, fail y release) en la misma actualización atómica que cambia el estado.

El mismo journal puede servir además como backend de durabilidad de un bot (eventsDurability({ journal }) de @botmesh/events/durability, ver Procesamiento durable de mensajes). Los registros entrantes del bot se escriben en su propio canal ('botwa:durable'). list() y replay() los muestran, pero un EventWorker nunca los reclama: un claim sin channel solo toma registros que no tienen canal. El propio bot los procesa.

Inbox y CRM

@botmesh/inbox incorpora asignación con concurrencia optimista (compare-and-set sobre version), handoff, notas y unread. Proyecta los StoredMessage del ledger con inbox.projectMessage(message) para reconstruir conversaciones entre canales; attachInbox(bot, inbox) lo hace para un bot en marcha, tanto con los mensajes entrantes como con los envíos aceptados, sin retrasar ninguno; un fallo de la proyección va al evento 'error' del bot ({ error }) o, si nadie lo escucha, al logger del bot. Sin la opción tenantId, las conversaciones van al tenant del propio bot (BotConfig.tenantId). Persiste con new Inbox(new KvInboxStore(kv)). Las notas se paginan con listNotes(tenant, conversación, { limit, after }) (after = el id de la última nota de la página anterior), e inbox.erase(tenant, conversación) borra una conversación con sus notas (derecho de supresión; borra también los mensajes en el store del bot).

@botmesh/crm guarda canales de contacto, tags, atributos y consentimiento de marketing, con segmentos por reglas (tagsAll, atributos comparados de forma estricta, consentimiento). ContactDirectory es async sobre MemoryContactStore o KvContactStore; setConsent reintenta a los escritores concurrentes en vez de pisarlos, y erase(tenant, id) borra un contacto (derecho de supresión). Segmenta siempre las campañas con datos aislados por tenant y con consentimiento.

Campañas, plugins y despliegues

@botmesh/campaigns descarta opt-outs, segmentos que no coinciden y destinatarios duplicados. El store reserva un hueco de envío por (tenant, campaña, destinatario) con TTL; markSent cuenta el envío y markFailed libera el hueco. sendCampaign(bot, planner, …) envía con bot.broadcast() y asienta cada delivery con el resultado real (aceptado, fallido, descartado por un hook o cancelado), emparejado por el chat de destino aunque el proveedor responda con un id canónico; si el propio broadcast rechaza, se liberan todos los huecos reservados.

@botmesh/hub valida manifiestos de plugins, compatibilidad con el runtime y permisos concedidos antes de habilitar un plugin. Los manifiestos se tratan como JSON no confiable (se comprueba el tipo de cada campo y se rechazan los permisos desconocidos) y el enable vuelve a comprobar los permisos de forma atómica, así que una reinstalación concurrente con más permisos nunca queda habilitada sin ellos. Las instalaciones y el enable/disable se persisten por tenant; por sí solo no ejecuta ni aísla código no confiable.

@botmesh/deployments registra versiones inmutables y un despliegue activo por tenant/bot, con reservas para gates en ejecución y auditoría de gate runs. list(tenant, bot, { limit, after }) pagina las versiones, prune(tenant, bot, keep) borra las anteriores a las keep más nuevas (nunca la activa) y new DeploymentRegistry(store, { retainVersions }) poda tras cada publicación, así que la lista de versiones no crece sin límite. Conéctalo a tu sistema de despliegue para hacer rollouts graduales y rollback físico.

Los cinco comparten el mismo modelo de almacenamiento: Memory…Store (referencia), Kv…Store sobre cualquier DurableKv, una suite de conformidad en test-utils/ y aislamiento de tenant impuesto por el nombre de la tabla en el backend (@botmesh/tenant-kv). Toda escritura se valida antes de llegar al backend (un campo mal formado falla con un TypeError que lo nombra), así que una escritura mala nunca deja ilegibles los listados de un tenant.

Trazas (OpenTelemetry)

bot.metrics() cuenta; las trazas explican. Con BotConfig.telemetry el bot emite un árbol de spans por mensaje (entrada → dispatch → handler / turno de conversación → llamadas al modelo y herramientas de IA → envío), con los ids con hash por defecto y sin contenido de mensajes. @botmesh/otel manda esos spans a OpenTelemetry (Jaeger, Grafana Tempo, cualquier backend OTLP) y publica bot.metrics() como métricas de OpenTelemetry (bot.plug(otelMetricsPlugin())). Configuración, catálogo de spans, política de datos personales y un ejemplo con Jaeger / Grafana Tempo: Trazas.