Eventos e Observabilidade

O Kikwiflow expõe dois canais de eventos completamente independentes, com garantias e propósitos diferentes. Confundir os dois é o erro de design mais comum ao integrar observabilidade — escolha o canal errado e você acaba com telemetria que promete consistência que não entrega, ou com um outbox transacional pesado demais para simples métricas de performance.

	Eventos leves (lightweight)	Eventos críticos (outbox)
Flag de ativação	kikwiflow.stats.enabled	kikwiflow.outbox.events-enabled
Entrega	Fire-and-forget, assíncrona, em memória	Persistida na mesma transação do UnitOfWork (implementação MongoDB)
Garantia	Nenhuma — pode se perder em crash/restart	Durável — commit atômico junto com a mudança de estado
Uso recomendado	Métricas de performance, dashboards, APM	Auditoria/histórico de execução, e como base para um consumidor de relay (ver nota sobre escopo abaixo)

:::info Escopo: esta lib entrega a persistência, não o relay O Kikwiflow OSS garante que o evento crítico existe e é consultável — de forma transacional e durável. A drenagem/entrega desses eventos para um sistema externo (Kafka, webhook, outro banco) — o "R" do padrão outbox-relay — é deliberadamente deixada como um ponto de extensão (kikwi-critical-event-relay-api), sem uma implementação embutida nesta versão. O schema da coleção outbox_events (campos relayStatus/lockedUntil, descritos abaixo) já foi desenhado para suportar um consumidor de relay construído pela sua equipe, sem precisar de migração de dados. :::

Eventos leves: ExecutionEventListener

A interface pública que uma aplicação implementa para observar a execução é:

@FunctionalInterface
public interface ExecutionEventListener {
    void onEvents(List<LightweightEvent> events);
}

public interface LightweightEvent {
    Instant getTimestamp();
}

Qualquer bean Spring que implemente ExecutionEventListener é automaticamente registrado no AsynchronousEventPublisher pelo KikwiflowAutoConfiguration (via ObjectProvider<List<ExecutionEventListener>>) — não é preciso fiação manual:

@Component
public class MetricsListener implements ExecutionEventListener {
    @Override
    public void onEvents(List<LightweightEvent> events) {
        events.forEach(event -> {
            if (event instanceof SyncContinuationFailed failure) {
                log.warn("Continuação síncrona falhou: instância={}, nó={}, erro={}",
                        failure.processInstanceId(), failure.failedNodeDefinitionId(), failure.errorMessage());
            }
        });
    }
}

AsynchronousEventPublisher: entrega via pool de virtual threads

public class AsynchronousEventPublisher implements EventPublisher {
    private final List<ExecutionEventListener> listeners = new CopyOnWriteArrayList<>();
    private final ExecutorService listenerExecutor; // Executors.newVirtualThreadPerTaskExecutor()

    public void publishEvents(List<LightweightEvent> events) {
        for (ExecutionEventListener listener : listeners) {
            listenerExecutor.submit(() -> listener.onEvents(events));
        }
    }
}

Cada listener é notificado em uma submissão de executor separada — um listener lento ou que lança exceção não bloqueia nem afeta outros listeners, nem a execução do processo em si, já que a publicação acontece de forma totalmente desacoplada do caminho crítico de persistência.

:::danger Sem garantia de entrega SyncContinuationFailed, o único evento leve emitido pelo motor (disparado quando uma continuação síncrona falha após a conclusão de uma tarefa externa — ver KikwiflowEngine.completeExternalTask), é publicado via fire-and-forget. Se a JVM cair entre a falha e a entrega do evento ao listener, o evento é perdido silenciosamente. Nunca implemente lógica de negócio essencial (ex.: reverter um efeito colateral, notificar um cliente) dentro de um ExecutionEventListener — use o outbox para isso. :::

O catálogo de eventos críticos: CriticalEventType

Todo evento crítico conhecido pelo motor está registrado em um único enum, usado tanto para construir eventos (em vez de literais de string soltos pelo código) quanto para o dispatch polimórfico na camada de persistência:

public enum CriticalEventType {
    FLOW_NODE_FINISHED(FlowNodeFinished.class),
    GATEWAY_ANSWER_RESOLVED(GatewayAnswerResolved.class),
    PROCESS_INSTANCE_FINISHED(ProcessInstanceFinished.class),
    PROCESS_INSTANCE_STARTED(ProcessInstanceStarted.class),
    INCIDENT_CREATED(IncidentCreated.class),
    INCIDENT_RESOLVED(IncidentResolved.class),
    EXTERNAL_TASK_CLAIMED(ExternalTaskClaimed.class),
    EXTERNAL_TASK_UNCLAIMED(ExternalTaskUnclaimed.class),
    EXTERNAL_TASK_COMPLETED(ExternalTaskCompleted.class),
    RETRY_SCHEDULED(RetryScheduled.class),
    PROCESS_VARIABLE_CHANGED(ProcessVariableChanged.class),
    TIMER_FIRED(TimerFired.class),
    ORPHANED_CHILD_COMPLETION(OrphanedChildCompletion.class);

    private final Class<? extends CriticalEvent> payloadType;
}

public interface CriticalEvent {
    String processInstanceId();
    String processDefinitionId();
}

Toda implementação de CriticalEvent expõe processInstanceId()/processDefinitionId() — isso é o que permite à camada de persistência indexar e consultar eventos por instância sem precisar de uma cadeia de instanceof.

Eventos críticos: o padrão Outbox

Cada mudança de estado relevante do processo é serializada como um OutboxEventEntity — com identidade própria (id, gerado automaticamente) — e incluída no mesmo UnitOfWork transacional que persiste a mudança de estado em si:

public class OutboxEventEntity {
    private String id;          // UUID gerado no construtor — vira o _id do documento persistido
    private Instant timestamp;
    private String event;       // nome do CriticalEventType, ex. "FLOW_NODE_FINISHED"
    private CriticalEvent payload;

    public OutboxEventEntity(CriticalEventType type, CriticalEvent payload) { /* ... */ }
}

Quando kikwiflow.outbox.events-enabled=true, a implementação MongoDB (MongoKikwiEngineRepository.commitWork) grava cada OutboxEventEntity do UnitOfWork na coleção outbox_events, dentro da mesma transação nativa (ClientSession.withTransaction) que persiste a mudança de estado da instância:

if (outboxPersistenceEnabled && unitOfWork.events() != null && !unitOfWork.events().isEmpty()) {
    List<InsertOneModel<Document>> eventWrites = unitOfWork.events().stream()
            .map(OutboxEventMapper::toDocument)
            .map(InsertOneModel::new)
            .toList();
    outboxEvents.bulkWrite(clientSession, eventWrites);
}

Isso significa que, ao contrário dos eventos leves, um evento crítico nunca é perdido silenciosamente: ou a transação inteira (mudança de estado + evento) é commitada, ou nenhuma das duas é — consistência garantida pelo mesmo mecanismo transacional descrito em Persistência e Consistência Transacional.

:::info A checagem da flag acontece de novo, no ponto de persistência kikwiflow.outbox.events-enabled é verificada tanto na construção do evento quanto, independentemente, dentro de commitWork, imediatamente antes da escrita física na coleção. Isso garante que "outbox desabilitado" signifique de fato "nada é escrito", mesmo que algum caminho upstream decida montar o UnitOfWork com eventos preenchidos. :::

Tipos de evento crítico

Todo evento crítico é montado dentro do motor, que checa kikwiflow.outbox.events-enabled num único ponto — com o outbox desligado, nada é construído nem gravado.

event	Payload (CriticalEvent)	Quando é emitido
FLOW_NODE_FINISHED	FlowNodeFinished	Ao final de cada execução de nó — sucesso, erro ou interrupção (boundary timer/error handler). Inclui nodeExecutionStatus (SUCCESS/INTERRUPTED/ERROR), flowNodeName/flowNodeDescription, startedAt/finishedAt, interruptedByNodeDefinitionId quando aplicável, e errorType/errorMessage/errorStackTrace quando nodeExecutionStatus == ERROR.
GATEWAY_ANSWER_RESOLVED	GatewayAnswerResolved	Sempre que um EXCLUSIVE_GATEWAY resolve uma resposta — carrega providerType, providerBean/providerVariable, a resposta resolvida e o id da aresta escolhida. Ver Navegação e Gateways de Decisão.
PROCESS_INSTANCE_FINISHED	ProcessInstanceFinished	Quando uma instância atinge ProcessInstanceStatus.COMPLETED. Carrega o snapshot final de variáveis, businessValue, tenantId, origin, parentInstanceId/callerTaskId/callerBranchId (call activity) e getDurationMs() derivado de startedAt/endedAt.
PROCESS_INSTANCE_STARTED	ProcessInstanceStarted	Contraparte de PROCESS_INSTANCE_FINISHED — emitido uma única vez, na primeira continuação da instância (ProcessStarter.execute()). Carrega o mesmo tipo de snapshot inicial (variáveis, businessValue, tenantId, origin), sem endedAt/status. actorId vem de ProcessStarter.byActor(...) — null se não informado.
INCIDENT_CREATED	IncidentCreated	Quando FailureHandler.handleFailure abre um incidente (retries esgotados, ou ProcessErrorException não tratado por nenhum boundary). actorId é sempre IdentityContext.system().
INCIDENT_RESOLVED	IncidentResolved	Quando KikwiflowEngine.retryIncident resolve manualmente um incidente aberto. actorId vem do IdentityContext do chamador.
EXTERNAL_TASK_CLAIMED	ExternalTaskClaimed	Quando KikwiflowEngine.claim atribui uma external/human task a um assignee. actorId vem do IdentityContext do chamador — normalmente igual a assignee, mas podem divergir (ex.: um supervisor atribuindo em nome de outro).
EXTERNAL_TASK_UNCLAIMED	ExternalTaskUnclaimed	Quando KikwiflowEngine.unclaim remove o assignee de uma external/human task — carrega previousAssignee e actorId (do IdentityContext do chamador).
EXTERNAL_TASK_COMPLETED	ExternalTaskCompleted	Quando KikwiflowEngine.completeExternalTask completa a tarefa. Carrega assignee (quem estava atribuído no momento, null se nunca foi clamada) e actorId (quem comandou o complete) lado a lado — o motor não valida um contra o outro, o complete é soberano; esse par existe só para que quem consome o evento detecte "completado por alguém diferente do atribuído" (ver nota abaixo).
RETRY_SCHEDULED	RetryScheduled	Quando FailureHandler.handleFailure reagenda uma tarefa que falhou mas ainda tem retries disponíveis (o ramo que não abre incidente). actorId é sempre IdentityContext.system().
PROCESS_VARIABLE_CHANGED	ProcessVariableChanged	Um evento por variável alterada via KikwiflowEngine.setVariables ou removida via KikwiflowEngine.unsetVariables. No caso de remoção, o payload traz removed=true e value=null; caso contrário, removed=false e o valor bruto — masking é responsabilidade de quem consome o outbox, não deste ponto de construção (ver nota sobre escopo/masking abaixo). actorId vem do IdentityContext do chamador.
TIMER_FIRED	TimerFired	Quando um timer não-interruptivo dispara e se reagenda (ContinuationService, ramo NON_INTERRUPTIVE_TIMER). Timers interruptivos são cobertos por FLOW_NODE_FINISHED(INTERRUPTED). actorId é sempre IdentityContext.system().
ORPHANED_CHILD_COMPLETION	OrphanedChildCompletion	Quando uma instância filha de um CALL_ACTIVITY_COORDINATOR conclui, mas a coordenadora do pai já não existe mais (cancelada por timeout antes). processInstanceId/processDefinitionId/tenantId identificam a instância filha — não o pai. Construído pela camada de persistência. Sem actorId — nunca é uma ação de usuário.

:::info Quem comandou a ação: actorId e o sentinel IdentityContext.system() 9 dos 13 tipos carregam actorId. Para os que nascem de uma chamada explícita de um usuário/API (claim, unclaim, completeExternalTask, retryIncident, setVariables, ProcessStarter.byActor), actorId vem do IdentityContext.actorId() do chamador. Para os que o motor dispara sozinho, sem nenhum humano por trás (IncidentCreated, RetryScheduled, TimerFired — todos originados em FailureHandler/ContinuationService reagindo a uma falha ou um timer), actorId é sempre o sentinel IdentityContext.system().actorId() ("__KIKWIFLOW_SYSTEM__"). FLOW_NODE_FINISHED, GATEWAY_ANSWER_RESOLVED e PROCESS_INSTANCE_FINISHED não carregam actorId — eles nascem dentro da execução do fluxo, onde nem sempre há uma identidade para propagar (ex.: um nó retomado por um worker em segundo plano). :::

:::info EXTERNAL_TASK_COMPLETED: soberania preservada, auditoria separada assignee/actorId divergentes em EXTERNAL_TASK_COMPLETED não é um erro — completeExternalTask sempre completa a tarefa, atribuída ou não, para quem quer que a chame (esse é um requisito de produto: o complete não pode depender de estar clamada). O evento só existe para que consumidores de auditoria/observabilidade possam perguntar "isso foi completado por alguém diferente do atribuído?" sem que o motor precise saber ou se importar com a resposta. :::

:::info tenantId está em todos os 13 tipos CriticalEvent.tenantId() é obrigatório em todos os payloads — dá para filtrar o histórico por tenant sem exceções. :::

Todos os payloads carregam tanto processDefinitionId (o id interno, específico da versão implantada) quanto, quando disponível, processDefinitionKey (o identificador estável usado em startProcess().byKey(...), o mesmo entre versões) — para agrupar/filtrar execuções de "o mesmo processo" ao longo de reimplantações, use processDefinitionKey.

Todos os tipos só são construídos quando kikwiflow.outbox.events-enabled está ligado — kikwiflow.stats.enabled não afeta o outbox.

:::info claim/unclaim/setVariables/unsetVariables não passam por UnitOfWork Diferente da maioria dos eventos (que viajam dentro de um UnitOfWork já existente), EXTERNAL_TASK_CLAIMED, EXTERNAL_TASK_UNCLAIMED e PROCESS_VARIABLE_CHANGED (tanto o set quanto o unset) são construídos em KikwiflowEngine e passados diretamente para KikwiEngineRepository.claim/unclaim/addVariables/unsetVariables — métodos que hoje mutam estado fora do padrão UnitOfWork/commitWork. A implementação MongoDB mesmo assim garante atomicidade: cada um desses métodos abre sua própria ClientSession.withTransaction para gravar a mudança de estado e o evento crítico juntos. :::

:::warning Masking na escrita: nunca. Na leitura via API: sempre PROCESS_VARIABLE_CHANGED grava value sem aplicar VariableSecurityPolicyManager.applyReadPoliciesAndMasking — decisão de escopo deliberada: o motor não decide o que é sensível no momento de persistir o evento. Isso não significa que ninguém aplica masking: a API REST de histórico (GET /process-instances/{id}/events, abaixo) aplica o policy manager no momento da leitura, antes de devolver a resposta. Um relay/consumer que leia outbox_events diretamente (sem passar por essa API) ainda recebe o valor bruto — é responsabilidade desse consumidor aplicar sua própria política. :::

API REST de histórico: GET /process-instances/{id}/events

kikwi-management-rest expõe a timeline completa de uma instância — a mesma lista de findEventHistoryByProcessInstanceId, mapeada para HistoryEventSummary (kikwi-model, io.kikwiflow.model.event):

public record HistoryEventSummary(
        String id,
        CriticalEventType eventType,
        String processInstanceId,
        String processDefinitionId,
        String tenantId,
        String actorId,
        Instant timestamp,
        CriticalEvent payload
) {}

HistoryEventSummaryMapper (kikwi-management-rest) promove ao topo os campos comuns (para não obrigar o cliente a fazer instanceof só para montar uma linha de timeline) e mantém o payload completo, específico do tipo, para o detalhamento. Quando payload é ProcessVariableChanged, o mapper aplica VariableSecurityPolicyManager.applyReadPoliciesAndMasking antes de devolver — se a policy negar leitura da variável, value vem como null na resposta, nunca o bruto.

:::info Sem enforcement de tenant neste endpoint — de propósito Diferente do que seria natural supor, este endpoint não valida identityContext.tenantId() contra o tenantId da instância. É uma decisão consciente: este endpoint é para uso backoffice/suporte/TI (o "Cockpit" do Kikwiflow), não uma API de uso geral de aplicação cliente — quem acessa o monitor já tem permissão independente de tenant. Aplicações que expõem esse dado para usuários finais devem implementar seu próprio enforcement de tenant na camada de aplicação, não esperar que o Kikwiflow faça isso aqui. :::

Registro do endpoint é condicional a kikwiflow.history.enabled (default true, KikwiflowHistoryProperties, módulo kikwi-management-rest-spring-boot-autoconfigure) — desligar essa flag remove o endpoint inteiro, mesmo com kikwiflow.outbox.events-enabled=true (que controla se o dado existe, não se é servido). 404 se a instância não existir; lista vazia (não erro) se o outbox nunca foi habilitado para essa instância.

Reconstruindo por onde uma instância passou

A coleção outbox_events não é apenas uma fila de relay transiente — entradas nunca são deletadas quando confirmadas/relayadas (apenas marcadas), então ela dobra como histórico durável e consultável de execução. QueryRepository.findEventHistoryByProcessInstanceId(String processInstanceId) retorna todos os eventos de uma instância, ordenados por timestamp ascendente:

List<OutboxEventEntity> history = kikwiEngineRepository.findEventHistoryByProcessInstanceId(processInstanceId);
// ex.: CONSULTA_BUREAU (FLOW_NODE_FINISHED) -> CALCULAR_SCORE (FLOW_NODE_FINISHED)
//      -> GATEWAY_FAIXA_RISCO (GATEWAY_ANSWER_RESOLVED, resposta "ALTO") -> ...

Isso só retorna dados quando kikwiflow.outbox.events-enabled=true — sem persistência habilitada, a lista é sempre vazia. Note que apagar uma instância concluída (deleteProcessInstanceById) não apaga seu histórico em outbox_events — é uma decisão consciente de design: o histórico sobrevive à instância.

Retenção: TTL nativo do MongoDB

Por padrão, outbox_events retém tudo indefinidamente. Se isso não for desejável (auditoria de curto prazo, custo de armazenamento, ou o consumo é feito por uma ferramenta de relay que já move os dados para outro lugar), configure kikwiflow.outbox.ttl com uma duração ISO-8601:

kikwiflow:
  outbox:
    events-enabled: true
    ttl: P30D   # eventos expiram 30 dias após o timestamp — MongoDB TTL nativo (expireAfterSeconds)

Isso cria (ou ajusta, via collMod, se o valor mudar entre deploys) um índice TTL real sobre o campo timestamp — a expiração é feita pelo próprio MongoDB em background, sem nenhum job/scheduler do Kikwiflow envolvido. Se ttl não for declarado, nenhum índice TTL é criado e a retenção é indefinida (comportamento padrão). Diminuir ou remover o TTL depois de já criado não apaga o índice automaticamente — isso exige um passo manual (db.outbox_events.dropIndex("outbox_ttl_idx")).

Ponto de extensão para relay

O Kikwiflow não traz um relay embutido: nenhum processo drena outbox_events para fora em segundo plano. O schema, porém, já suporta um consumidor construído pela sua equipe — cada documento tem relayStatus (PENDING/LOCKED/RELAYED) e lockedUntil, pensados para um claim atômico via findOneAndUpdate (o mesmo padrão de Workers e Tarefas Externas), com o índice relay_status_timestamp_idx já criado. A interface OutboxReader (kikwi-critical-event-relay-api) define o contrato esperado (readAndLockNextBatch/confirmBatch), sem implementação fornecida.

Escolhendo o canal certo

:::tip Regra prática

• Precisa de um dashboard de latência por nó, sem garantias fortes? Eventos leves.
• Precisa de auditoria durável, ou consultar por onde uma instância específica passou (depuração, suporte)? Outbox, via findEventHistoryByProcessInstanceId.
• Precisa entregar esses eventos para um sistema externo (Kafka, webhook, data warehouse)? Outbox como base de dados de origem — mas você (ou uma lib de relay dedicada) precisa implementar o consumidor; esta versão não inclui um pronto para uso.
• Só precisa consultar o estado atual de uma instância (não seu histórico)? Nenhum dos dois — use o lado de query do CQRS (QueryRepository/ExternalTaskQueryService), descrito na Visão Geral. :::