Coreografia entre serviços

Como os 7 serviços internos coreografam via Redis Streams — a visão dinâmica, orientada a eventos, em cenários concretos.

Este guia é o complemento em movimento de outros dois:

Guia	Foco	O que responde
Arquitetura do sistema	Visão estática	O quê existe — serviços, packages, streams, caches.
Coreografia entre serviços (este)	Visão dinâmica	Como os serviços se coordenam em sequência — quem publica o quê, em resposta a quê.
Redis consumers	Infra de consumo	Como cada consumer se conecta — conexão bloqueante dedicada, política de ACK/retry.

Nenhum serviço interno chama outro via HTTP. Toda coordenação acontece por publicação em stream + consumo por consumer group. Não há orquestrador central que comanda os outros: cada serviço reage ao que aparece no seu stream e publica o próximo evento. Isso é coreografia, não orquestração — daí o nome.

Para o ciclo de vida de ponta a ponta de uma conversa (do primeiro "oi" ao ticket fechado) numa narrativa única, veja Jornada do usuário. Este guia foca a mecânica dos streams.

Mapa completo produtor → consumidor(grupo)

Fonte da verdade: STREAMS e GROUPS em packages/flow-ai-redis/src/constants.ts; o pareamento stream↔grupo em STREAM_GROUP_MAP (packages/flow-ai-redis/src/streams.ts).

A constante STREAMS tem 18 nomes, mas apenas 17 streams vivas (com pelo menos um consumer group registrado). A 18ª (OUTGOING) é legado inerte — está listada por último, fora do mapa.

#	Stream	Produtor(es)	Consumidor(es) — `grupo` (serviço)	Payload
1	`stream:incoming`	meta-api, ig-api, webchat-gateway	`orchestrator` (orchestrator)	`IncomingStreamEvent` — mensagem do usuário (sempre em shape WhatsApp)
2	`stream:flow`	orchestrator, engine, agent, core	`engine` (engine)	`FlowStreamEvent` — `userInput` / `humanSessionEnded` / `agentSessionEnded` / `externalJump`
3	`stream:helpdesk`	orchestrator, engine, agent, core	`core` (core)	Handoff, mensagens em atendimento humano, tickets de IA
4	`stream:outgoing-meta`	engine, agent, core	`send` (meta-api) + `persist` (orchestrator)	Mensagem a enviar via WhatsApp
5	`stream:outgoing-chat`	engine, agent, core	`webchat` (webchat-gateway) + `persist` (orchestrator)	Mensagem a entregar no browser
6	`stream:outgoing-ig`	engine, core, ig-api	`send` (ig-api) + `persist` (orchestrator, registrado mas não lido)	Mensagem a enviar via Instagram
7	`stream:status`	meta-api, ig-api, webchat-gateway	`status` (orchestrator) + `core-status` (core) + `core-campaigns-status` (core)	Status de entrega (`sent`/`delivered`/`read`/`failed`)
8	`stream:wa-flow-logs`	meta-api	`core-wa-flow-logs` (core) + `core-wa-flow-deferred` (core)	Logs de webhook de WhatsApp Flows
9	`stream:templates`	meta-api	`core-templates` (core)	Eventos WABA-level de status/quality de template
10	`stream:analytics`	engine	`core-analytics` (core)	Eventos `recordEvent` → `event_occurrences`
11	`stream:tool-executions`	engine	`core-tool-executions` (core)	Execuções de Tool → `tool_execution_log`
12	`stream:agent`	engine, agent, orchestrator, core	`agent` (agent)	`AgentStreamPayload` — início/retomada ou `manualHandoff`
13	`stream:llm-usage`	agent	`core-llm-usage` (core)	Uso de tokens/custo → `llm_usage`
14	`stream:agent-turns`	agent	`core-agent-turns` (core)	Turnos do agente (AI Debugger) → `agent_turns`
15	`stream:summary`	engine, agent, core	`core-summary` (core)	Pedidos de sumarização via LLM
16	`stream:ig-comments`	ig-api	`ig-comments` (ig-api)	Comentários de post/live IG (moderação)
17	`stream:ig-comment-activations`	ig-api	`core-ig-comment-activations` (core)	Disparos de gatilho de comentário → `instagram_comment_activations`
—	`stream:outgoing`	(nenhum)	(nenhum)	Legado/inativo — constante `OUTGOING`, fora do `STREAM_GROUP_MAP`

Três pontos de precisão que o mapa revela e que valem destaque:

Fan-out de saída: outgoing-meta e outgoing-chat têm dois grupos concorrentes-não-competidores — persist (durabilidade no Postgres, via orchestrator) e send/webchat (entrega ao canal). A mesma entry é entregue uma vez a cada grupo, permitindo persistir e enviar em paralelo.
send cobre dois streams: o grupo send está registrado tanto em outgoing-meta quanto em outgoing-ig, mas é consumido por serviços diferentes (meta-api e ig-api, respectivamente) — cada processo só lê o stream do seu canal.
status tem três grupos: status (orchestrator persiste o status da mensagem), core-status (core reemite via Socket.io ao desk) e core-campaigns-status (core atualiza CampaignRecipient).

Nota — stream:outgoing é legado/inativo. A constante OUTGOING existe em constants.ts por referência, mas não está no STREAM_GROUP_MAP e não tem produtor nem consumidor. Os três streams de saída vivos são outgoing-meta, outgoing-chat e outgoing-ig. Docs antigas (e vários CLAUDE.md) ainda dizem stream:outgoing — trate como drift.

Nota — saída de Instagram não é persistida. O grupo persist está registrado em outgoing-ig (STREAM_GROUP_MAP), mas nenhum consumidor o lê. Consequência: mensagens de saída do Instagram são enviadas pelo send (ig-api), porém não gravadas no Postgres pelo orchestrator.

Como escolher o stream de saída

Os serviços de runtime (engine, agent, core) não decidem o canal por config — deduzem do prefixo do sessionId. A função resolveContentChannel(sessionId) (services/flow-ai-engine/src/runtime/publish.ts) mapeia:

Prefixo do `sessionId`	Canal	Stream de saída
`ig-*`	Instagram	`stream:outgoing-ig`
`wc-*`	WebChat	`stream:outgoing-chat`
(qualquer outro, ex. `wa-`)*	WhatsApp	`stream:outgoing-meta`

Nota — o flow-ai-agent não roteia para outgoing-ig. O agent só publica em outgoing-meta / outgoing-chat (services/flow-ai-agent/src/tools/response.ts). Uma sessão de agente de IA respondendo numa conversa ig-* publica em stream:outgoing-meta (canal incorreto). É drift conhecido — documentado, não corrigido.

O padrão comum: `runStreamLoop`

Todo consumer de stream do sistema roda dentro do helper runStreamLoop (packages/flow-ai-redis/src/shutdown.ts). Ele concentra o que é crítico e propenso a erro — conexão bloqueante, checagem de shutdown e sinalização de saída — para que essa lógica viva num lugar só, não espalhada por ~20 loops. Cada consumer passa apenas o onBatch, que contém o parse/xack/telemetria específicos daquele stream.

// services/flow-ai-orchestrator/src/consumer.ts — forma canônica de um consumer
export function consumeIncoming(): Promise<void> {
  return runStreamLoop({
    group: GROUPS.ORCHESTRATOR,
    consumer: CONSUMER_NAME,
    streams: [STREAMS.INCOMING],
    log: logger,
    label: "orchestrator",
    onBatch: async (results) => {
      for (const { messages } of results) {
        for (const { id, fields } of messages) {
          await runWithMessageContext(STREAMS.INCOMING, GROUPS.ORCHESTRATOR, id, fields, async () => {
            // 1. parse + validação de shape → xack de descarte se inválido
            // 2. handler de negócio
            // 3. xack só quando a entry pode sair da pending list
          })
        }
      }
    },
  })
}

O que runStreamLoop faz por dentro (shutdown.ts):

Cria a conexão bloqueante dedicada com createBlockingClient() (redis.duplicate()) — cada loop com XREADGROUP ... BLOCK segura sua própria conexão TCP, para não atrasar XADD/XACK/SET no singleton compartilhado. Ver Redis consumers para o porquê da conexão dedicada.
Registra-se no drain (registerLoop()) — o shutdown espera este loop sair antes de fechar recursos.
Entra no while (!isShuttingDown()): lê um batch com xreadgroup(group, consumer, streams, count=10, blockMs=2000, client), chama onBatch, repete.
Envelopa onBatch num try/catch externo: um batch que rejeita (ex.: um xack de descarte que lança) não pode derrubar o consumer — loga, espera 1s e segue; entries não-ackadas caem no PEL e são recuperadas depois.

Consumer groups: distribuição vs. fan-out

Um consumer group distribui as entries de um stream entre os consumers daquele grupo (cada entry vai para um consumer só, com garantia at-least-once via PEL). Dois grupos diferentes no mesmo stream recebem cada um uma cópia de toda entry — é o fan-out que sustenta o persist + send.

setupStreams() (streams.ts) cria todas as streams (MKSTREAM) e todos os grupos (XGROUP CREATE ... $) no boot de cada serviço, ignorando BUSYGROUP em restarts. É idempotente e deve rodar antes de qualquer xreadgroup.

Detalhe importante: stream:agent roda com um único consumer (CONSUMER_NAME = "agent-1") no grupo agent, que processa uma entry por vez — o consumo é globalmente serial, não há paralelismo dentro do agent.

ACK e retry — a política não é uniforme

Cada onBatch decide quando fazer xack. A regra geral (orchestrator e engine):

Caso	Ação	Racional
Entry sem `payload`	`xack` (descarta)	Não recuperável.
JSON inválido	`xack` (descarta)	Não recuperável.
Shape fora do contrato	`xack` (descarta)	Não corresponde ao tipo da stream.
Handler de negócio lança	sem `xack`	Entry fica no PEL → reentregue depois.
Erro de leitura do Redis	sem `xack`, sleep 1s	Falha fora de uma entry específica.

Mas o flow-ai-agent inverte o último caso. Em services/flow-ai-agent/src/consumer.ts o xack fica fora do try/catch do handler:

try {
  if (!("kind" in event)) await handle(event)
  else if (event.kind === "manualHandoff") await handleManualHandoff(event)
} catch (error) {
  logger.error({ entryId: id, err: error }, "[flow-ai-agent] erro ao processar evento")
}
await xack(STREAMS.AGENT, GROUPS.AGENT, id) // ← acka MESMO se o handler lançou

Ou seja: um turno de agente que falha é ackado assim mesmo — não há reentrega via PEL. Na prática, o efeito é que um turno LLM que lançou não é reprocessado (o que evitaria regastar tokens e duplicar respostas); o erro é apenas logado e a sessão segue no próximo input do usuário. Contraste com orchestrator/engine, onde um handler que lança deixa a entry pendente para retry.

Ao documentar reentrega, seja específico por serviço: "entry fica pendente no PEL" vale para orchestrator e engine, não para o agent.

Graceful shutdown — `onPreDrain` / `onPostDrain`

O shutdown ordenado vive em packages/flow-ai-redis/src/shutdown.ts e é instalado uma vez por serviço com installShutdownHandlers({ service, log, drainTimeoutMs }). O problema que resolve: sem drenagem, um SIGINT/SIGTERM (o pm2 manda SIGINT) mataria o processo no meio de um batch, deixando entries sem xack presas no PEL.

Ordem determinística de teardown ao receber o sinal:

SIGINT / SIGTERM
   │
   ▼
1. isShuttingDown() = true        → loops param de pegar batch NOVO
   │                                (o BLOCK corrente retorna em até blockMs,
   │                                 a flag é vista, o loop sai limpo)
   ▼
2. onPreDrain                     → para de aceitar trabalho novo
   │                                (Fastify app.close(), socket.io io.close())
   │                                NÃO afeta o processamento de stream em voo
   ▼
3. espera loopExits               → cada loop termina o batch atual, faz xack e sai
   │                                (bounded por drainTimeoutMs)
   ▼
4. onPostDrain                    → libera recursos do processamento
   │                                (ex.: prisma.$disconnect())
   ▼
5. quitBlockingClients()          → fecha conexões bloqueantes (sem reads em voo)
   ▼
6. closeSubscriberConnection()    → fecha pub/sub de invalidação
   ▼
7. quitRedis()                    → Redis compartilhado por ÚLTIMO (xack/xadd usam ele)
   │
   ▼
process.exit(0)

Hook	Quando roda	Registrar aqui	Não registrar aqui
`onPreDrain(label, fn)`	Antes de esperar os loops drenarem	Recursos que param de aceitar trabalho novo — `app.close()` (Fastify), `io.close()` (Socket.io)	Recursos usados no processamento em voo
`onPostDrain(label, fn)`	Depois dos loops drenarem	Recursos usados durante o processamento — `prisma.$disconnect()`	O Redis compartilhado (fechado por último, automaticamente)

Por que essa ordem importa para a coreografia: um serviço em shutdown termina o batch em voo e o acka antes de fechar o Redis — então nenhum evento fica meio-processado. Um watchdog força process.exit(1) se o drain exceder drainTimeoutMs + 5s; o kill_timeout do pm2 precisa ser maior que isso, senão um SIGKILL interrompe o drain.

Loops de polling (não-stream) usam runPollingLoop + interruptibleDelay, que acordam imediatamente no shutdown para não segurar o drain com um sleep longo.

Cenário 1 — inbound → flow → outbound

O caminho mais comum: usuário manda mensagem, o bot responde. Três serviços, quatro streams.

usuário (WhatsApp)
   │  webhook HTTP
   ▼
flow-ai-meta-api ──── xadd stream:incoming ───►  flow-ai-orchestrator
                                                      │  consome (grupo orchestrator)
                                                      │  persiste Message, resolve/atualiza SessionState
                                                      │  mode === "flow"
                                                      │
                              xadd stream:flow ◄──────┘  (FlowStreamEvent kind:"userInput")
                                     │
                                     ▼
                              flow-ai-engine
                                     │  consome (grupo engine)
                                     │  carrega SessionState + PublishedFlow, roda a máquina de estados
                                     │  resolveContentChannel("wa-...") → outgoing-meta
                                     │
        ┌────────────────────────────┴────────────────────────────┐
        │ xadd stream:outgoing-meta (fan-out: DOIS grupos)          │
        ▼                                                            ▼
  grupo persist (orchestrator)                             grupo send (meta-api)
  grava Message pending no Postgres                        envia à Meta Cloud API
                                                           │  publica status
                                                           ▼
                                             xadd stream:status ──► orchestrator (persist status)
                                                                 └─► core (Socket.io → desk)

Pontos-chave:

O orchestrator roteia por SessionState.mode (services/flow-ai-orchestrator/src/handlers/incoming.ts): flow → xadd(STREAMS.FLOW), agent → xadd(STREAMS.AGENT), human → xadd(STREAMS.HELPDESK).
O engine escolhe o stream de saída pelo prefixo do sessionId — aqui wa-* → outgoing-meta.
outgoing-meta faz fan-out: persist grava no Postgres, send envia à Meta. Grupos distintos, cada um recebe a entry.
O status de entrega volta por stream:status — três grupos o consomem (persistir, reemitir ao desk, atualizar campanha).

Detalhe cross-service: os quatro variants de FlowStreamEvent (userInput, humanSessionEnded, agentSessionEnded, externalJump) chegam todos no mesmo grupo engine — o parseFlowEvent do engine (services/flow-ai-engine/src/consumer.ts) discrimina pelo campo kind.

Cenário 2 — handoff → helpdesk → ticket

Quando o flow entra num HumanAttendanceBlock (ou o agent decide transferir), a sessão passa para atendimento humano.

flow-ai-engine (flow chega no HumanAttendanceBlock)
   │  SessionState.mode = "human"; setSession()
   │
   │  xadd stream:helpdesk (HelpdeskHandoffEvent)
   ▼
flow-ai-core
   │  consome (grupo core, startHelpdeskConsumer)
   │  cria Ticket (kind:"human", openReason:"human_handoff", status:"waiting")
   │  aplica regra de distribuição da fila
   │  emite via Socket.io: helpdesk:waiting, ticket novo ao agente
   ▼
flow-ai-desk (agente vê o ticket)

...enquanto mode === "human", cada nova mensagem do usuário:
usuário ──► meta-api ──► stream:incoming ──► orchestrator (mode=human)
                                                 └─ xadd stream:helpdesk ──► core ──► Socket.io ──► desk

...respostas do agente saem pelo core:
core ── xadd stream:outgoing-meta ──► fan-out persist + send ──► Meta

Pontos-chave:

O handoff pode nascer em três produtores de stream:helpdesk: engine (bloco de atendimento), orchestrator (roteamento com mode=human já persistido — xadd(STREAMS.HELPDESK) em incoming.ts) e agent (transferência via tool, ou ticket de IA).
Enquanto mode === "human", o orchestrator não roteia para o engine — cada input do usuário vai direto ao stream:helpdesk.
O flow-ai-core é o único consumidor de stream:helpdesk (grupo core) e o único que fala Socket.io com o desk. Ver Atendimento humano e Tickets do helpdesk para o ciclo de vida do ticket.
Uma sessão travada por erro de runtime permanece em mode: "human" com frozenByError — não existe modo frozen.

Cenário 3 — AgentBlock → stream:agent → agentSessionEnded → retorno ao flow

O caso mais coreografado: o flow cede o controle a um agente de IA, que roda um loop LLM e depois devolve o controle ao flow.

flow-ai-engine (flow chega num AgentBlock)
   │  session.mode = "agent"  (runtime/execute.ts)
   │  monta AgentStreamEvent (SEM campo kind); setSession() ANTES de publicar
   │
   │  xadd stream:agent  (handlers/flow.ts: result.agentEvent)
   ▼
flow-ai-agent
   │  consome (grupo agent, consumer serial "agent-1")
   │  parse: "kind" ausente → handle() ; kind:"manualHandoff" → handleManualHandoff()
   │  roda o loop LLM (Responses API): monta contexto, chama tools, encadeia agentes
   │
   │  durante o loop, publica nos side-channels:
   │     xadd stream:agent-turns  (AI Debugger)
   │     xadd stream:llm-usage    (tokens/custo)
   │     xadd stream:summary      (resumo)
   │     xadd stream:helpdesk     (cria/fecha ticket de IA, kind:"ai")
   │     xadd stream:outgoing-meta / -chat  (respostas ao usuário)
   │
   │  enquanto mode === "agent", cada input do usuário:
   │     orchestrator (mode=agent) ── xadd stream:agent ──► agent (retomada)
   │
   ▼  o agente chama endAgentSession (tools/actions.ts)
   │  monta AgentSessionEndedEvent (kind:"agentSessionEnded")
   │
   │  xadd stream:flow
   ▼
flow-ai-engine
   │  consome (grupo engine); parseFlowEvent vê kind:"agentSessionEnded"
   │  sai do AgentBlock avaliando outputConditions, retoma walkBlocks
   ▼
flow segue no bloco seguinte (mode volta a "flow")

Pontos-chave:

O AgentStreamEvent de início não tem campo kind — é o discriminante que o parseFlowEvent/consumer do agent usa. O único variant com kind em stream:agent é manualHandoff (AgentStreamPayload = AgentStreamEvent | AgentManualHandoffEvent).
Ordem obrigatória no engine: setSession() antes do xadd(STREAMS.AGENT) — o comentário em execute.ts marca o agentEvent como "deve ser publicado APÓS setSession()", senão o agent leria uma sessão sem mode:"agent".
A retomada (input do usuário com mode=agent) e o início trafegam no mesmo stream:agent, no mesmo grupo — o agent trata os dois no handle().
O fim do loop é uma volta ao stream:flow com kind:"agentSessionEnded" — o mesmo stream do Cenário 1, discriminado por kind. O engine avalia só as outputConditions do AgentBlock e segue.

Nota — canal errado no IG. Como o agent nunca publica em outgoing-ig, um agente respondendo numa sessão ig-* (Cenário 3) manda a resposta por outgoing-meta. Drift conhecido — ver seção "Como escolher o stream de saída".

Para o interior do loop LLM (multi-agente, agentChain, tools, pré-processamento de mídia), veja Agentes de IA.

Cenário 4 — side-channels de telemetria

Além do fluxo principal, o engine e o agent alimentam streams fire-and-forget consumidos só pelo flow-ai-core, que os persiste. Não há resposta ao usuário — são trilhas de auditoria, métrica e observabilidade.

flow-ai-engine ──► stream:analytics        ──► core (core-analytics)        ──► event_occurrences
flow-ai-engine ──► stream:tool-executions  ──► core (core-tool-executions)  ──► tool_execution_log
flow-ai-agent  ──► stream:agent-turns      ──► core (core-agent-turns)      ──► agent_turns
flow-ai-agent  ──► stream:llm-usage        ──► core (core-llm-usage)        ──► llm_usage
engine/agent/  ──► stream:summary          ──► core (core-summary)          ──► resumo via LLM → Chat/Ticket/Contact
core

gateways       ──► stream:status           ──► orchestrator (status)        ──► Message.status
                                            ├─► core (core-status)          ──► Socket.io → desk
                                            └─► core (core-campaigns-status)──► CampaignRecipient.status

Stream	Produtor	Grupo (core)	Destino no Postgres
`stream:analytics`	engine	`core-analytics`	`event_occurrences`
`stream:tool-executions`	engine	`core-tool-executions`	`tool_execution_log`
`stream:agent-turns`	agent	`core-agent-turns`	`agent_turns` (AI Debugger, retenção 30d)
`stream:llm-usage`	agent	`core-llm-usage`	`llm_usage`
`stream:summary`	engine, agent, core	`core-summary`	resumo gerado via LLM, gravado em `Chat`/`Ticket`/`Contact`
`stream:status`	meta-api, ig-api, webchat-gateway	`status`, `core-status`, `core-campaigns-status`	`Message.status`, Socket.io ao desk, `CampaignRecipient`

Por que streams e não gravação direta: desacoplam o caminho crítico (executar o flow, responder ao usuário) da escrita analítica. Se o core estiver sobrecarregado, os eventos ficam no stream e são drenados depois — sem travar o engine nem o agent.

O trace atravessa a coreografia

Toda essa coreografia — inclusive os side-channels — vive num único trace de OpenTelemetry, mesmo saltando entre serviços por Redis. O xadd injeta o traceparent W3C numa cópia dos campos da mensagem (packages/flow-ai-redis/src/streams.ts), e cada consumer reconstrói o contexto com runWithMessageContext(...) no onBatch (visível nos três consumers). Sem telemetria ativa, o propagador global é no-op e o carrier fica idêntico — custo zero. Ver Logging para a config de observabilidade.

Resumo das transições de stream

De	Evento	Para	Discriminante
orchestrator	input, `mode=flow`	`stream:flow` (`userInput`)	`SessionState.mode`
orchestrator	input, `mode=agent`	`stream:agent`	`SessionState.mode`
orchestrator	input, `mode=human`	`stream:helpdesk`	`SessionState.mode`
engine	resposta do bot	`stream:outgoing-{meta,chat,ig}`	prefixo do `sessionId`
engine	`HumanAttendanceBlock`	`stream:helpdesk`	tipo do bloco
engine	`AgentBlock`	`stream:agent` (`AgentStreamEvent`)	tipo do bloco
agent	`endAgentSession`	`stream:flow` (`agentSessionEnded`)	tool chamada
agent	resposta LLM	`stream:outgoing-{meta,chat}`	prefixo do `sessionId`
core	fecha ticket humano	`stream:flow` (`humanSessionEnded`)	ação do agente
core	redirect externo	`stream:flow` (`externalJump`)	API de sessão

Arquivos relevantes

Arquivo	Papel
`packages/flow-ai-redis/src/constants.ts`	`STREAMS` (18 nomes) e `GROUPS` — fonte da verdade dos nomes.
`packages/flow-ai-redis/src/streams.ts`	`xadd` (injeta `traceparent`), `xreadgroup`, `xack`, `setupStreams`, `STREAM_GROUP_MAP` (17 streams vivas).
`packages/flow-ai-redis/src/shutdown.ts`	`runStreamLoop`, `runPollingLoop`, `installShutdownHandlers`, `onPreDrain`, `onPostDrain`, `interruptibleDelay`.
`packages/flow-ai-redis/src/client.ts`	`createBlockingClient` (conexão bloqueante dedicada), `quitBlockingClients`, `quitRedis`.
`services/flow-ai-orchestrator/src/consumer.ts`	Consumer de `stream:incoming`; parse + política de ACK padrão.
`services/flow-ai-orchestrator/src/handlers/incoming.ts`	Roteamento por `mode` → `flow` / `agent` / `helpdesk`.
`services/flow-ai-engine/src/consumer.ts`	Consumer de `stream:flow`; discrimina os 4 variants de `FlowStreamEvent`.
`services/flow-ai-engine/src/handlers/flow.ts`	Publica `agentEvent` em `stream:agent` após `setSession()`.
`services/flow-ai-engine/src/runtime/execute.ts`	`session.mode = "agent"` no `AgentBlock`; saída do bloco no `agentSessionEnded`.
`services/flow-ai-engine/src/runtime/publish.ts`	`resolveContentChannel(sessionId)` — escolha do stream de saída por prefixo.
`services/flow-ai-agent/src/consumer.ts`	Consumer de `stream:agent`; `xack` fora do try/catch (sem retry via PEL).
`services/flow-ai-agent/src/tools/actions.ts`	`endAgentSession` → `xadd(STREAMS.FLOW)` (`agentSessionEnded`).

Veja também

Arquitetura do sistema — visão estática dos serviços, streams e caches.
Redis consumers — infra de conexão bloqueante e ACK/retry por consumer.
Jornada do usuário — ciclo de vida de ponta a ponta de uma conversa.
Fluxo inbound completo e Fluxo outbound completo — os dois primeiros saltos em detalhe.
Agentes de IA — o interior do loop LLM do Cenário 3.

Coreografia entre serviços ​

Mapa completo produtor → consumidor(grupo) ​

Como escolher o stream de saída ​

O padrão comum: runStreamLoop ​

Consumer groups: distribuição vs. fan-out ​

ACK e retry — a política não é uniforme ​

Graceful shutdown — onPreDrain / onPostDrain ​

Cenário 1 — inbound → flow → outbound ​

Cenário 2 — handoff → helpdesk → ticket ​

Cenário 3 — AgentBlock → stream:agent → agentSessionEnded → retorno ao flow ​

Cenário 4 — side-channels de telemetria ​

O trace atravessa a coreografia ​

Resumo das transições de stream ​

Arquivos relevantes ​

Veja também ​

Coreografia entre serviços

Mapa completo produtor → consumidor(grupo)

Como escolher o stream de saída

O padrão comum: `runStreamLoop`

Consumer groups: distribuição vs. fan-out

ACK e retry — a política não é uniforme

Graceful shutdown — `onPreDrain` / `onPostDrain`

Cenário 1 — inbound → flow → outbound

Cenário 2 — handoff → helpdesk → ticket

Cenário 3 — AgentBlock → stream:agent → agentSessionEnded → retorno ao flow

Cenário 4 — side-channels de telemetria

O trace atravessa a coreografia

Resumo das transições de stream

Arquivos relevantes

Veja também