import type { ServiceBoundary } from "@/lib/atendimento/fronteira";
/**
 * Node handlers for the follow-up flow engine (Task 4.1) — PURE, no DB access.
 * `engine.ts` owns the tick/DB orchestration; this file only decides "given
 * this node + these facts, what happens next" so it's testable without Postgres.
 */
import { NO_REPLY_BRANCH_ID, REPEAT_BODY_BRANCH_ID, REPEAT_DONE_BRANCH_ID, nodeBranches } from "./graph-schema";
import type { FlowEdge, FlowNode, ReplySaveTo } from "./graph-schema";
import { avaliarDataRelativa, ehOpDeData, type OpDeData } from "./condicao-por-data";
import { parseReplyCount } from "./parse-count";
import { janelaDaEspera, primeiroInstanteNoHorario } from "./espera-no-horario";
import { clampEspera, esperaPlanejadaDe, type EsperaAdaptativa } from "./timing-plan";
import { fraseDeConfirmacao } from "./vocabulario";

export type EnrollmentStatus =
  | "active"
  | "waiting_reply"
  /**
   * Espera longa imune à resposta (nó `wait` com `immune_to_reply`).
   *
   * TEM relógio como `active` — é o `next_eval_at` que a acorda —, mas está
   * fora de `LIVE_STATUSES` em `reactivity.ts`, então a mensagem do contato não
   * a cancela nem corta o timer, e fora do índice único anti-spam, então o
   * contato continua podendo entrar noutra cadência enquanto dorme.
   */
  | "dormente"
  | "paused_handoff"
  /**
   * Pausa pedida por uma pessoa (0145, `STATUS_PAUSA_MANUAL` em `intervencao.ts`).
   * Sem relógio e fora de `LIVE_STATUSES`: a mensagem comum não a acorda — quem a
   * retoma é quem pausou. O opt-out a alcança (`reactivity.ts`). Faltava neste
   * tipo desde a 0145, e foi por isso que a lista do opt-out pôde esquecê-la sem
   * o compilador ter como apontar.
   */
  | "paused_manual"
  /**
   * Roteiro de atendimento em andamento (0394). Conduzido pelo TURNO, não pelo
   * relógio: o motor de follow-up nunca o reclama (o claim filtra
   * `active|waiting_reply`). Está aqui porque o opt-out o alcança
   * (`reactivity.ts`) e o cancelamento pela fila o encerra.
   */
  | "coletando"
  | "completed"
  | "cancelled"
  | "dead";

export type EnrollmentOutcome = "converted" | "replied" | "exhausted" | "opted_out" | "handoff";

/**
 * Snapshot of a `followup_enrollments` row — plain data (not tied to any DB
 * client) so both the pg-backed test adapter and a future supabase-js adapter
 * can produce it. Field names mirror the table (migration 0054) 1:1.
 */
export interface EnrollmentRow {
  service_boundary?: ServiceBoundary | null;
  revision?: number;
  appointment_id?: string | null;
  appointment_revision?: number | null;
  id: string;
  organization_id: string;
  pointer_id: string;
  version_id: string;
  contact_id: string;
  conversation_id: string | null;
  current_node_id: string;
  status: EnrollmentStatus;
  next_eval_at: string | null;
  claimed_until: string | null;
  attempts: number;
  max_attempts: number;
  last_error: string | null;
  steps_taken: number;
  outcome: EnrollmentOutcome | null;
  cancel_reason: string | null;
  started_at: string;
  completed_at: string | null;
  updated_at: string;
  /**
   * Plano de tempo decidido no acionamento (migration 0144) — `unknown` porque
   * é `jsonb` e um clone pode ter qualquer coisa lá; quem lê é
   * `esperaPlanejadaDe` (timing-plan.ts), que degrada em vez de lançar.
   * Ausente/`null` = enrollment de antes da feature ⇒ comportamento anterior.
   */
  timing_plan?: unknown;
}

/** Minimal typed facts a `condition` node can check — loaded by the engine, never guessed. */
export interface LeadFacts {
  /**
   * Fuso da ORGANIZAÇÃO, para a condição por data comparar dia de calendário —
   * ver `condicao-por-data.ts` — e para a espera com horário (`ate_o_horario`,
   * `espera-no-horario.ts`). Ausente cai no padrão do produto; o motor só o
   * resolve quando o grafo tem regra de data ou espera com horário.
   */
  fuso?: string | null;
  lead_stage: string | null;
  tags: string[];
  steps_taken: number;
  /**
   * Desfecho do passo anterior — a classe que o último `ai_classify` escolheu,
   * lida dos eventos da inscrição (`ultimoDesfechoDe`). `null` quando o fluxo
   * ainda não classificou nada; e `null` NÃO satisfaz `neq` (ver `evaluateCheck`).
   */
  last_outcome: string | null;
  contact_name?: string | null;
  custom_fields?: Record<string, unknown>;
}

/** Reference to a `followup_enrollment_events` row — only what `resolveWaitPhase` needs. */
export interface EnrollmentEventRef {
  node_id: string | null;
  idempotency_key: string | null;
  event_type?: string | null;
  payload?: Record<string, unknown> | null;
}

export type NodeResult =
  // `reason` só aparece quando o avanço NÃO é o avanço comum: hoje, o trigger
  // desistindo do plano de tempo (o turno nunca voltou). Vira event_type próprio
  // no engine — seguir sem plano é um fato que o operador precisa poder ler.
  //
  // `class` só aparece quando o avanço É uma classificação decidida pelo motor:
  // o `ai_classify` que sai por "sem resposta" porque a carência venceu. Vai para
  // o payload do evento e é o que `ultimoDesfechoDe` lê — o desfecho não pode
  // depender de QUEM tirou o lead do classificar.
  | { kind: "advance"; next_node_id: string; next_eval_at: Date; reason?: "plan_timeout"; repeat?: { index: number; total: number }; class?: string }
  // stays on the node. `wake_status` parks `match_reply` in waiting_reply without a job.
  | { kind: "wait"; next_eval_at: Date; wake_status?: "active" | "waiting_reply" | "dormente" }
  | {
      kind: "enqueue_turn";
      purpose: "send_message" | "classify" | "plan_timing";
      wake_status: "active" | "waiting_reply";
      fixed_body?: string;
    }
  // action recheck: the send turn is already in flight; stay put WITHOUT re-enqueuing (anti-dup-send).
  | { kind: "recheck"; next_eval_at: Date }
  // action dead-man: the turn never completed after MAX_ACTION_RECHECKS — give up (engine routes to markDead).
  | { kind: "dead"; reason: string }
  // outcome is nullable for the 'custom' end-node case (cancel_reason carries the note instead).
  | { kind: "complete"; outcome: EnrollmentOutcome | null; cancel_reason?: string }
  | { kind: "fail"; error: string };

/** Backoff ladder indexed by `attempts - 1` (clamped to the last slot) — 30s..1h. */
export const BACKOFF_MS = [30_000, 60_000, 300_000, 900_000, 3_600_000] as const;

/** Recheck cadence while an action's send turn is in flight — how long the engine waits before
 *  looking again to see if the turn landed. Imported by engine.ts for the enqueue next_eval_at too. */
export const ACTION_RECHECK_MS = 5 * 60_000;

/** Teto do backoff entre rechecks — a partir daqui a espera não cresce mais. */
export const ACTION_RECHECK_MAX_MS = 60 * 60_000;

/**
 * Dead-man bound: idle rechecks tolerated on an action node before a turn that never completes
 * (worker down / permanently failing) is markDead — never re-enqueues, never waits forever.
 *
 * ⚠️ ERA 5, E 5 × 5min MATAVA TODO FOLLOW-UP DA NOITE. A espera do envio tem um
 * motivo LEGÍTIMO e longo que este contador não distinguia de "worker morto": a
 * janela anti-ban (7h–22h no padrão). Um toque que caísse às 22h ficava ~25 min
 * em recheck e o enrollment morria com `action_turn_never_completed` — o lead
 * nunca recebia, e o motivo registrado era falso. Medido em produção
 * (2026-08-18): enrollment `dead` no nó de abertura, com o worker vivo e o turno
 * apenas esperando a janela.
 *
 * Com o backoff de `atrasoDoRecheck`, este orçamento cobre ~11h — mais que a
 * maior noite fechada — e ainda custa poucos ticks. O dead-man continua
 * existindo: worker realmente morto termina em `dead`, só que depois de uma
 * espera que não confunde noite com defeito.
 *
 * ⚠️ E SUBIR O NÚMERO NÃO É A DEFESA — a defesa é `EVENTO_ACAO_ADIADA`.
 * Aumentar o teto só compra tempo contra a espera mais longa que alguém
 * configurou, e essa espera não tem teto: as horas e os dias da janela
 * anti-ban são knobs por canal (uma noite de sábado com domingo fechado já dá
 * 33h), e a faixa de envio do agente permite um único dia da semana (159h).
 * Contra um orçamento fixo, esse jogo não se ganha. O que o resolve é o turno
 * DIZER que está estacionado, e o contador medir só a ociosidade depois disso
 * — ver `rechecksOciososDaAcao` logo abaixo.
 */
export const MAX_ACTION_RECHECKS = 14;

/**
 * Quanto esperar até o próximo recheck da ação: 5min dobrando até 1h.
 *
 * Exponencial e não fixo porque as duas causas de espera têm escalas
 * diferentes: turno em voo volta em segundos (os primeiros rechecks são
 * curtos), janela fechada volta em horas (e aí não faz sentido perguntar de 5
 * em 5 minutos por 9 horas).
 */
export function destinoJaPreenchido(lead: LeadFacts, saveTo: ReplySaveTo): boolean {
  if (saveTo.kind === "contact_name") return Boolean(lead.contact_name?.trim());
  const v = lead.custom_fields?.[saveTo.key];
  if (typeof v === "string") return v.trim().length > 0;
  return v !== undefined && v !== null && v !== "";
}

/** Resposta que MANTÉM o valor já gravado no modo `confirm`. */
export function ehConfirmacao(body: string): boolean {
  const t = body.trim().toLowerCase();
  return /^(sim|s|yes|ok|isso|correto|confirmo|confirmar|pode)([.!]?)$/.test(t) || t === "isso mesmo";
}

function modoSeJaExiste(node: Extract<FlowNode, { type: "match_reply" }>): "skip" | "overwrite" | "confirm" {
  return node.config.if_exists ?? "overwrite";
}

export function atrasoDoRecheck(rechecksJaFeitos: number): number {
  const passo = Math.max(0, rechecksJaFeitos);
  return Math.min(ACTION_RECHECK_MS * 2 ** passo, ACTION_RECHECK_MAX_MS);
}

/**
 * O evento que o turno grava quando o envio foi ADIADO para um instante CONHECIDO
 * — janela fechada (anti-ban, ou a faixa do próprio agente), e não defeito.
 *
 * É PROVA DE VIDA, e essa é a razão de ele existir. O dead-man da ação mede
 * "rechecks sem o turno fechar", e essa medida não distingue duas situações
 * opostas: o worker morreu, e o worker está vivo e o envio está estacionado
 * até a janela abrir. Enquanto o adiamento era silencioso, as duas só se
 * pareciam — e o orçamento de ~11h de `MAX_ACTION_RECHECKS` era gasto por
 * espera legítima, matando o enrollment com um motivo falso
 * (`action_turn_never_completed`) enquanto o envio ainda ia acontecer.
 */
export const EVENTO_ACAO_ADIADA = "action_deferred";

/**
 * O turno de classificar rodou e o cliente ainda não tinha respondido ao envio
 * do fluxo — o nó segue esperando até a carência. Sem esta linha, o dossiê
 * mostrava "Pediu ao agente para interpretar a resposta" e mais nada por até a
 * carência inteira: parecia travado. Não é passo (a chave não é `${nó}:${passo}`),
 * então nenhum guarda de ocupação do motor a conta.
 */
export const EVENTO_CLASSIFICACAO_ESPERANDO = "classify_waiting";

/**
 * Rechecks ociosos da ação NESTA estadia — o número que o dead-man deve medir.
 *
 * Idêntico a `occupancyEventCount` enquanto não houver adiamento (o dead-man
 * continua exatamente tão severo com worker morto quanto antes); a diferença é
 * que ele PARA no último `action_deferred`. Cada adiamento é uma prova de vida
 * nova, e o que se conta é a ociosidade DEPOIS dela.
 */
export function rechecksOciososDaAcao(events: EnrollmentEventRef[], nodeId: string): number {
  let n = 0;
  for (let i = events.length - 1; i >= 0; i--) {
    const evento = events[i]!;
    if (evento.node_id !== nodeId) break;
    if (evento.event_type === EVENTO_ACAO_ADIADA || evento.event_type === EVENTO_TURNO_DESCARTADO) return n;
    n++;
  }
  return n;
}

/**
 * O turno de envio desta estadia saiu da fila SEM rodar: a organização foi
 * suspensa e `fn_org_parada_descarta_fila` (migration 0501) o falhou, gravando
 * este evento. Não é defeito do worker, então não conta para o dead-man (ver
 * `rechecksOciososDaAcao`), e o motor enfileira um turno novo na reativação —
 * o claim não entrega a inscrição enquanto a org está parada.
 */
export const EVENTO_TURNO_DESCARTADO = "turn_discarded";

/** O último turno desta estadia no `action` foi descartado e nenhum outro o substituiu. */
export function turnoDaAcaoDescartado(events: EnrollmentEventRef[], nodeId: string): boolean {
  for (let i = events.length - 1; i >= 0; i--) {
    const evento = events[i]!;
    if (evento.node_id !== nodeId) return false;
    if (evento.event_type === EVENTO_TURNO_DESCARTADO) return true;
    if (evento.event_type === "turn_enqueued") return false;
  }
  return false;
}

/**
 * Dead-man do PLANO de tempo: rechecks tolerados no `trigger` esperando o turno
 * de planejamento voltar. Menor que o da ação (3 × 5min ≈ 15min) e com desfecho
 * OPOSTO — aqui o fluxo SEGUE sem plano, nunca morre. Um planejador de tempo
 * indisponível não pode matar o follow-up: sem ele o fluxo ainda funciona
 * inteiro (cai no máximo de cada espera, que é o comportamento anterior);
 * matar o enrollment trocaria uma degradação por uma perda.
 */
export const MAX_PLAN_RECHECKS = 3;

export type EdgeMatch =
  | { type: "always" }
  | { type: "class_match"; value: string }
  | { type: "cond_result"; value: boolean }
  | { type: "branch"; branch_id: string };

/**
 * Qual saída de um nó de classificação leva à classe `classe` — resolvendo os
 * dois dialetos (nó v1 casa por texto, nó migrado casa pelo id estável do ramo).
 *
 * Existe como função porque a MESMA pergunta é feita em dois pontos do caminho
 * de execução: aqui, quando o prazo vence sem resposta, e no `turn-bridge`,
 * quando o modelo classifica. Consertar só um dos dois deixava o fluxo migrado
 * roteando certo para quem responde e errado, em silêncio, para quem não
 * responde — que num follow-up é o caso mais comum. Achado pelo DevVivo na
 * revisão: eu tinha ensinado o `selectEdge` a casar ramo e usado isso só no
 * `condition`.
 *
 * Casa por rótulo E por id de propósito: `no_reply` é reservado (id `no_reply`,
 * rótulo "Sem resposta"), e uma classe do usuário é achada pelo texto que ele
 * escreveu.
 */
export function classEdgeMatch(
  node: Extract<FlowNode, { type: "ai_classify" | "match_reply" }>,
  classe: string,
): EdgeMatch {
  const ramo = nodeBranches(node).find(
    (b) => b.kind === "match" && (b.label === classe || b.id === classe),
  );
  return ramo?.condition.type === "branch"
    ? { type: "branch", branch_id: ramo.condition.branch_id }
    : { type: "class_match", value: classe };
}

/**
 * Picks the outbound edge from `from`: highest `priority` first, exact
 * condition match tried first, `always` as fallback. `null` if nothing fits.
 */
export function selectEdge(edges: FlowEdge[], from: string, match: EdgeMatch): FlowEdge | null {
  const candidates = edges.filter((e) => e.source === from).slice().sort((a, b) => b.priority - a.priority);

  const exact = candidates.find((e) => {
    switch (match.type) {
      case "always":
        return e.condition.type === "always";
      case "class_match":
        return e.condition.type === "class_match" && e.condition.value === match.value;
      case "cond_result":
        return e.condition.type === "cond_result" && e.condition.value === match.value;
      case "branch":
        return e.condition.type === "branch" && e.condition.branch_id === match.branch_id;
    }
  });
  if (exact) return exact;

  if (match.type !== "always") {
    const fallback = candidates.find((e) => e.condition.type === "always");
    if (fallback) return fallback;
  }
  return null;
}

/**
 * A `wait` node is entered twice: once to start the timer (writes the
 * generic step event), once after `next_eval_at` elapses to advance. Both
 * ticks see the SAME node (current_node_id unchanged) with `steps_taken`
 * incrementing by exactly 1 on every applied step (engine.ts) — so "did we
 * already start this wait" is exactly "does the event for the PRIOR step on
 * this node exist".
 */
export function occupancyEventCount(events: EnrollmentEventRef[], nodeId: string): number {
  let n = 0;
  for (let i = events.length - 1; i >= 0; i--) {
    if (events[i]!.node_id !== nodeId) break;
    n++;
  }
  return n;
}

/**
 * O turno de envio desta estadia no `action` já fechou (`action_sent`).
 * Se o enrollment ainda aponta pro action, foi corrida com `action_recheck`
 * (ou update perdido no completeTurn) — o motor deve avançar, não rechecar.
 */
export function actionTurnCompleted(events: EnrollmentEventRef[], nodeId: string): boolean {
  for (let i = events.length - 1; i >= 0; i--) {
    if (events[i]!.node_id !== nodeId) break;
    if (events[i]!.event_type === "action_sent") return true;
  }
  return false;
}

export function repeatTakenFromEvents(events: EnrollmentEventRef[], nodeId: string): number {
  return events.filter(
    (e) => e.node_id === nodeId && typeof e.payload?.repeat_index === "number",
  ).length;
}

export function repeatTotalFromEvents(events: EnrollmentEventRef[], nodeId: string): number | null {
  for (let i = events.length - 1; i >= 0; i--) {
    const total = events[i]!.payload?.repeat_total;
    if (events[i]!.node_id === nodeId && typeof total === "number") return total;
  }
  return null;
}

export function latestRepeatIndex(events: EnrollmentEventRef[]): { index: number; total: number } | null {
  for (let i = events.length - 1; i >= 0; i--) {
    const index = events[i]!.payload?.repeat_index;
    const total = events[i]!.payload?.repeat_total;
    if (typeof index === "number" && typeof total === "number") return { index, total };
  }
  return null;
}

export function resolveWaitPhase(events: EnrollmentEventRef[], nodeId: string, stepsTaken: number): boolean {
  const priorKey = `${nodeId}:${stepsTaken - 1}`;
  return events.some((e) => e.node_id === nodeId && e.idempotency_key === priorKey);
}

/**
 * Piso do inbound que casa neste `match_reply`: o instante em que a espera
 * começou, não o `updated_at` da inscrição.
 *
 * O `inbound_woke` (e qualquer tick depois) regrava `updated_at`. Usar essa
 * coluna como piso esconde a mensagem que ACORDOU a espera — ela chegou
 * segundos antes do wake. `wait_started.payload.next_eval_at` é park+graça,
 * então park = next_eval_at − grace_timeout_ms.
 */
export function pisoDoInboundDaEspera(
  node: Extract<FlowNode, { type: "match_reply" }>,
  events: EnrollmentEventRef[],
  fallback: string,
): string {
  for (let i = events.length - 1; i >= 0; i--) {
    const e = events[i]!;
    if (e.node_id !== node.id) continue;
    if (e.event_type !== "wait_started") continue;
    const next = e.payload?.next_eval_at;
    if (typeof next !== "string") break;
    const start = Date.parse(next) - node.config.grace_timeout_ms;
    if (Number.isFinite(start)) return new Date(start).toISOString();
    break;
  }
  return fallback;
}

/**
 * Passos é número, mas o formulário gravou por meses o que se DIGITAVA — texto.
 * Com `"3"`, `gte` nunca era verdadeiro e `neq` sempre era: a regra aparecia
 * pronta no card e decidia sozinha. Lê o número que a pessoa escreveu; texto que
 * não é número segue como está (e o publish o recusa).
 */
function valorDePassos(value: string | number): string | number {
  if (typeof value === "number") return value;
  const limpo = value.trim();
  const n = Number(limpo);
  return limpo !== "" && Number.isFinite(n) ? n : value;
}

/**
 * Os eventos que gravam a classe com que o lead SAIU de um `ai_classify` — a
 * fonte do "Desfecho do passo anterior". São dois escritores e um só campo
 * (`payload.class`):
 *   - `ai_classified`: a ponte, quando o modelo classificou a resposta;
 *   - `node_advanced` com `class`: o motor, quando a carência venceu sem
 *     resposta e o lead saiu por `no_reply` (`case "ai_classify"` abaixo).
 * Até o turno de classificar parar de concluir `no_reply` sozinho, só o
 * primeiro existia; sem o segundo, "Sem resposta" deixava o desfecho `null` (ou
 * o de uma volta anterior) e a condição mudava de ramo em silêncio.
 */
const EVENTOS_DE_DESFECHO = new Set(["ai_classified", "node_advanced"]);

/**
 * O desfecho do último passo que DECIDIU algo: a classe com que o lead saiu do
 * `ai_classify` mais recente da inscrição — pelo modelo ou pela carência vencida.
 * `null` quando ainda não houve classificação (fluxo que nunca passou por um
 * `ai_classify`, ou classificação que terminou sem classe).
 *
 * ⚠️ Este dado existia como CONTRATO (o rótulo "Desfecho do passo anterior" está
 * em `vocabulario.ts`, o campo está no enum do `graph-schema.ts` e a tela o
 * oferece) e não como dado: o motor montava `LeadFacts.last_outcome` como `null`
 * FIXO em `engine.ts`, então a condição escrita com ele era decorativa — o dono
 * da VPS montava o filtro e o follow-up ignorava. Era pior com `neq`, porque
 * `null !== "x"` é `true` e o fluxo mandava TODO lead pelo ramo da negativa.
 *
 * `events` chega na ordem do banco (`created_at` ascendente) — o ÚLTIMO evento de
 * classificação é o desfecho vigente, não importa quantos passos atrás ele ficou.
 */
export function ultimoDesfechoDe(events: EnrollmentEventRef[]): string | null {
  for (const evento of [...events].reverse()) {
    if (!EVENTOS_DE_DESFECHO.has(evento.event_type ?? "")) continue;
    const classe = evento.payload?.class;
    if (typeof classe === "string" && classe.length > 0) return classe;
  }
  return null;
}

function evaluateCheck(
  check: {
    field: "lead_stage" | "tag" | "steps_taken" | "last_outcome" | "lead_date";
    op: "eq" | "neq" | "gte" | "lte" | "contains" | OpDeData;
    value: string | number;
    key?: string;
  },
  lead: LeadFacts,
  agora: Date,
): boolean {
  // A regra de data sai ANTES de tudo: ela não tem valor "atual" comparável
  // pelos operadores comuns, e `custom_fields` é jsonb — qualquer coisa pode
  // estar lá. Campo ausente ou ilegível responde `false`, nunca `true`.
  if (check.field === "lead_date") {
    if (!ehOpDeData(check.op) || !check.key) return false;
    return avaliarDataRelativa({
      valor: lead.custom_fields?.[check.key],
      op: check.op,
      dias: check.value,
      agora,
      fuso: lead.fuso,
    });
  }
  // Operador de data em campo que não é de data: o schema recusa no publish, e
  // aqui responde `false` em vez de cair num `switch` que não o conhece.
  if (ehOpDeData(check.op)) return false;

  const actual: string | number | null | string[] =
    check.field === "lead_stage" ? lead.lead_stage
    : check.field === "tag" ? lead.tags
    : check.field === "steps_taken" ? lead.steps_taken
    : lead.last_outcome;

  if (Array.isArray(actual)) {
    // 'tag' é multi-valorado: eq/contains viram "está entre as tags"; gte/lte não fazem sentido.
    const included = actual.includes(String(check.value));
    if (check.op === "eq" || check.op === "contains") return included;
    if (check.op === "neq") return !included;
    return false;
  }

  // ⚠️ Desconhecido não satisfaz NEGAÇÃO.
  //
  // Sem esta linha, `neq` comparava `null` com o valor e respondia `true` — ou
  // seja, "não foi X" valia para TODO lead, inclusive o que nunca foi
  // classificado. É a segunda metade do defeito do "Desfecho do passo anterior":
  // com o campo alimentado, o lead cujo `ai_classify` ainda não rodou (ou que
  // terminou sem classe) passaria por qualquer condição escrita como negação, e
  // o fluxo seguiria pelo ramo errado em silêncio.
  //
  // `eq` e `contains` já eram falsos com `null` — negar não pode ser a única
  // porta que a ausência de dado abre. Ausência não prova a negativa: um lead
  // sem classificação não é um lead "que não foi hot".
  if (actual === null) return false;
  const expected = check.field === "steps_taken" ? valorDePassos(check.value) : check.value;
  switch (check.op) {
    case "eq":
      return actual === expected;
    case "neq":
      return actual !== expected;
    case "gte":
      return typeof actual === "number" && typeof expected === "number" && actual >= expected;
    case "lte":
      return typeof actual === "number" && typeof expected === "number" && actual <= expected;
    case "contains":
      return typeof actual === "string" && typeof expected === "string" && actual.includes(expected);
  }
}

/**
 * Exportada para a revalidação imediatamente antes do envio
 * (`condicao-antes-do-envio.ts`): a mesma função, nunca uma segunda régua —
 * duas avaliações da mesma condição que pudessem divergir seriam pior que
 * nenhuma revalidação.
 */
export function evaluateCondition(
  config: Extract<FlowNode, { type: "condition" }>["config"],
  lead: LeadFacts,
  agora: Date,
): boolean {
  const results = config.checks.map((check) => evaluateCheck(check, lead, agora));
  return config.combinator === "and" ? results.every(Boolean) : results.some(Boolean);
}

/**
 * Pure per-node decision. `waitElapsed` is resolved by the engine (via
 * `resolveWaitPhase` against real events) BEFORE calling this — optional so
 * non-`wait`/`ai_classify` calls don't need to pass it. For `ai_classify` it
 * means "a classify turn was already enqueued for this occupancy of the node"
 * (same prior-step-event check as `wait`) — re-entering with it `true` means
 * EITHER `grace_timeout_ms` elapsed without a completed classification OR
 * reactivity (Task 5.2, `lib/followup/reactivity.ts`) woke the node early
 * because an inbound reply arrived. `wokeEarly` is the signal that
 * disambiguates the two (own marker event, distinct from the
 * `classify_enqueued` event `waitElapsed` checks): `true` re-enqueues a fresh
 * classify turn with the real reply instead of auto-advancing via `no_reply`.
 */
export function processNode(input: {
  node: FlowNode;
  edges: FlowEdge[];
  enrollment: EnrollmentRow;
  lead: LeadFacts;
  clock: () => Date;
  waitElapsed?: boolean;
  wokeEarly?: boolean;
  /** Last inbound `messages.body` for this contact/conversation — engine loads on `match_reply` + wokeEarly. */
  lastInboundBody?: string;
  /** action occupancy guard: a `turn_enqueued` event for THIS stay on the action node already
   *  exists (an entry/recheck happened before). Resolved by the engine via `resolveWaitPhase`
   *  — same prior-step-event check as `wait`. When true, the send turn is in flight: DON'T
   *  re-enqueue (a second job_id would bypass the send sink's (job_id,seq) dedup → dup message). */
  actionEnqueued?: boolean;
  /** action dead-man counter: number of events already accumulated on this action node — used to
   *  bound rechecks so a turn that never completes routes to `dead` instead of looping forever. */
  actionRecheckCount?: number;
  /** action: `action_sent` já existe nesta estadia — o envio fechou; avançar (sara corrida com recheck). */
  actionCompleted?: boolean;
  /** trigger: as esperas adaptativas do grafo pinado (`coletarEsperasAdaptativas`). Vazio/ausente
   *  ⇒ não há o que planejar e o acionamento NÃO paga uma chamada de modelo. */
  smartWaits?: EsperaAdaptativa[];
  /** trigger occupancy guard: um turno de planejamento para ESTA estadia no trigger já foi
   *  enfileirado. Mesmo check de evento-do-passo-anterior do wait/action (`resolveWaitPhase`). */
  planEnqueued?: boolean;
  /** trigger dead-man counter: eventos já acumulados no nó trigger — limita os rechecks para que
   *  um turno de planejamento que nunca volta siga SEM plano em vez de esperar para sempre. */
  planRecheckCount?: number;
  /** `repeat`: quantas voltas deste nó já saíram por `body` (eventos com repeat_index). */
  repeatTaken?: number;
  /** `repeat`: N armado na primeira visita; null = ainda precisa parsear lastInboundBody. */
  repeatTotal?: number | null;
  /** Próximo nó pela aresta `always` — a ação olha o `match_reply` seguinte para pular o envio. */
  proximo?: FlowNode | null;
}): NodeResult {
  const {
    node,
    edges,
    enrollment,
    clock,
    lead,
    waitElapsed,
    wokeEarly,
    lastInboundBody,
    actionEnqueued,
    actionRecheckCount,
    actionCompleted,
    smartWaits,
    planEnqueued,
    planRecheckCount,
    repeatTaken,
    repeatTotal,
    proximo,
  } = input;

  switch (node.type) {
    case "trigger": {
      const edge = selectEdge(edges, node.id, { type: "always" });
      if (!edge) return { kind: "fail", error: `trigger node "${node.id}" has no outbound edge` };

      // ACIONAMENTO: é aqui que o plano de tempo do fluxo inteiro é decidido, uma
      // única vez, antes do primeiro passo. Fluxo sem espera adaptativa e
      // enrollment que já tem plano seguem direto — nenhum custo de modelo, e o
      // comportamento de antes desta feature fica intacto.
      // `?? null` de propósito: a coluna chega `null` do banco e `undefined` de
      // um snapshot montado antes da migration 0144 — os dois querem dizer "sem
      // plano ainda", e tratar só um deles pularia o planejamento em silêncio.
      const precisaPlanejar = (smartWaits?.length ?? 0) > 0 && (enrollment.timing_plan ?? null) === null;
      if (!precisaPlanejar) {
        return { kind: "advance", next_node_id: edge.target, next_eval_at: clock() };
      }
      if (!planEnqueued) {
        return { kind: "enqueue_turn", purpose: "plan_timing", wake_status: "active" };
      }
      if ((planRecheckCount ?? 0) >= MAX_PLAN_RECHECKS) {
        // O turno de planejamento nunca voltou. Seguir sem plano (cada espera cai
        // no seu máximo) é a degradação certa — ver MAX_PLAN_RECHECKS.
        return { kind: "advance", next_node_id: edge.target, next_eval_at: clock(), reason: "plan_timeout" };
      }
      return { kind: "recheck", next_eval_at: new Date(clock().getTime() + ACTION_RECHECK_MS) };
    }

    case "wait": {
      // Resposta do lead corta a espera: o timer é teto (ninguém respondeu),
      // não um atraso obrigatório depois de cada envio. Vale também com horário
      // (`ate_o_horario`): ele segura o toque PROATIVO, não a resposta a quem
      // acabou de escrever.
      if (wokeEarly) {
        const edge = selectEdge(edges, node.id, { type: "always" });
        if (!edge) return { kind: "fail", error: `wait node "${node.id}" has no outbound edge after elapsing` };
        return { kind: "advance", next_node_id: edge.target, next_eval_at: clock() };
      }
      const janelaDoNo =
        node.config.mode === "fixed" ? janelaDaEspera(node.config.ate_o_horario, lead.fuso) : null;
      if (!waitElapsed) {
        // Adaptativo: o instante vem do plano decidido no acionamento. Sem plano
        // legível para ESTE nó (enrollment anterior à feature, fluxo v1, jsonb
        // corrompido), cai no máximo — que era o comportamento único até aqui.
        //
        // O clamp é REFEITO aqui, contra o nó, mesmo a ponte já tendo clampado ao
        // gravar: "quem decide o intervalo é o nó" só é invariante se valer na
        // LEITURA. `timing_plan` é jsonb num banco que o self-hoster administra —
        // uma linha editada à mão, ou um bug futuro que grave sem clampar,
        // prenderia o lead muito além do que o operador configurou na tela, e
        // ninguém veria. Custa uma comparação por espera.
        const planejada = node.config.mode === "smart" ? esperaPlanejadaDe(enrollment.timing_plan, node.id) : null;
        const durationMs =
          node.config.mode === "fixed"
            ? node.config.duration_ms
            : planejada === null
              ? node.config.max_ms
              : clampEspera(planejada.escolhido_ms, node.config.min_ms, node.config.max_ms).escolhido_ms;
        // Espera imune dorme: o status tira a inscrição do alcance da
        // reatividade (que decide por status, sem carregar o grafo) e libera o
        // slot único anti-spam enquanto ela espera. Quem a acorda continua sendo
        // o `next_eval_at` abaixo, pelo mesmo claim — não há segundo agendador.
        const imune = node.config.mode === "fixed" && node.config.immune_to_reply === true;
        // Horário da espera (`ate_o_horario`): o fim que cai de noite ou num dia
        // fechado vai para a próxima abertura JÁ AQUI, na entrada — a fila mostra
        // o instante verdadeiro do próximo passo, não as 24 h cruas.
        return {
          kind: "wait",
          next_eval_at: primeiroInstanteNoHorario(new Date(clock().getTime() + durationMs), janelaDoNo),
          ...(imune ? { wake_status: "dormente" as const } : {}),
        };
      }
      // E é conferido de novo na SAÍDA. O fim calculado na entrada não é a única
      // origem do relógio: retomar uma pausa devolve o restante, e adiar grava a
      // hora que a pessoa escolheu — os dois podem cair fora do horário. Fora
      // dele, a espera continua até a abertura (custa um tick, como um recheck).
      const agora = clock();
      const abertura = primeiroInstanteNoHorario(agora, janelaDoNo);
      if (abertura.getTime() > agora.getTime()) {
        const imune = node.config.mode === "fixed" && node.config.immune_to_reply === true;
        return { kind: "wait", next_eval_at: abertura, ...(imune ? { wake_status: "dormente" as const } : {}) };
      }
      const edge = selectEdge(edges, node.id, { type: "always" });
      if (!edge) return { kind: "fail", error: `wait node "${node.id}" has no outbound edge after elapsing` };
      return { kind: "advance", next_node_id: edge.target, next_eval_at: clock() };
    }

    case "condition": {
      if (node.config.branching === "per_check") {
        // "Uma saída por regra": a PRIMEIRA regra que passa manda, e a ordem da
        // lista é a precedência — a mesma ordem que o usuário vê no formulário.
        // Duas regras verdadeiras não podem sortear caminho; `combinator` não
        // é consultado aqui, porque nesse modo a regra não vota, ela roteia.
        const hitId = node.config.checks.find((c) => c.id !== undefined && evaluateCheck(c, lead, clock()))?.id;
        // Nenhuma regra passou -> o ramo obrigatório 'else', que na aresta é `always`.
        // `selectEdge` também cai nele quando o usuário deixou um ramo sem ligar:
        // sair pela saída de escape é ruim, ficar preso no nó é pior.
        const edge =
          hitId === undefined
            ? selectEdge(edges, node.id, { type: "always" })
            : selectEdge(edges, node.id, { type: "branch", branch_id: hitId });
        if (!edge) {
          return {
            kind: "fail",
            error: `condition node "${node.id}" has no edge for branch "${hitId ?? "else"}"`,
          };
        }
        return { kind: "advance", next_node_id: edge.target, next_eval_at: clock() };
      }
      const result = evaluateCondition(node.config, lead, clock());
      const edge = selectEdge(edges, node.id, { type: "cond_result", value: result });
      if (!edge) return { kind: "fail", error: `condition node "${node.id}" has no matching edge for result ${result}` };
      return { kind: "advance", next_node_id: edge.target, next_eval_at: clock() };
    }

    case "ai_classify": {
      if (!waitElapsed || wokeEarly) {
        // 1ª entrada (waitElapsed=false) OU reactivity acordou cedo com uma
        // resposta real (wokeEarly=true, mesmo com waitElapsed=true — o marker
        // de reactivity é o desempate): reenfileira classify. Nunca conta como
        // 'no_reply' quando existe reply de verdade em voo.
        return { kind: "enqueue_turn", purpose: "classify", wake_status: "waiting_reply" };
      }
      // grace_timeout_ms venceu sem turno de classificação concluído — classifica
      // como 'no_reply' SEM chamar o LLM (onda 5, critério 2); selectEdge já cai
      // no fallback 'always' se não houver aresta 'no_reply' explícita.
      const edge = selectEdge(edges, node.id, classEdgeMatch(node, NO_REPLY_BRANCH_ID));
      if (!edge) return { kind: "fail", error: `ai_classify node "${node.id}" has no edge for class "no_reply" (fallback also missing)` };
      return { kind: "advance", next_node_id: edge.target, next_eval_at: clock(), class: NO_REPLY_BRANCH_ID };
    }

    case "match_reply": {
      if (!waitElapsed && !wokeEarly) {
        if (node.config.save_to && destinoJaPreenchido(lead, node.config.save_to)) {
          const modo = modoSeJaExiste(node);
          if (modo === "skip") {
            const edge = selectEdge(edges, node.id, { type: "always" });
            if (!edge) {
              return { kind: "fail", error: `match_reply node "${node.id}" has no fallback edge to skip` };
            }
            return { kind: "advance", next_node_id: edge.target, next_eval_at: clock() };
          }
          if (modo === "confirm") {
            const valor =
              node.config.save_to.kind === "contact_name"
                ? (lead.contact_name ?? "").trim()
                : String(lead.custom_fields?.[node.config.save_to.key] ?? "").trim();
            return {
              kind: "enqueue_turn",
              purpose: "send_message",
              wake_status: "waiting_reply",
              fixed_body: fraseDeConfirmacao(
                valor,
                node.config.save_to.kind === "contact_name" ? "contact_name" : "lead_custom",
              ),
            };
          }
        }
        return {
          kind: "wait",
          next_eval_at: new Date(clock().getTime() + node.config.grace_timeout_ms),
          wake_status: "waiting_reply",
        };
      }
      if (wokeEarly) {
        const body = (lastInboundBody ?? "").trim().toLowerCase();
        // inbound_woke sem texto desta pergunta (piso excluiu o "." que
        // enfileirou o menu) NÃO é ALWAYS nem no_reply — senão o fluxo
        // dispara o cardápio inteiro no mesmo request.
        if (!body) {
          if (!waitElapsed) {
            return {
              kind: "wait",
              next_eval_at: new Date(clock().getTime() + node.config.grace_timeout_ms),
              wake_status: "waiting_reply",
            };
          }
        } else {
          const hit =
            node.config.save_to !== undefined
              ? undefined
              : node.config.branches.find((b) => {
                  const needle = b.pattern.trim().toLowerCase();
                  if (needle.length === 0) return false;
                  return b.op === "eq" ? body === needle : body.includes(needle);
                });
          const edge = hit
            ? selectEdge(edges, node.id, { type: "branch", branch_id: hit.id })
            : selectEdge(edges, node.id, { type: "always" }) ??
              (() => {
                const ramo = node.config.branches.find((b) => b.id !== NO_REPLY_BRANCH_ID);
                return ramo ? selectEdge(edges, node.id, { type: "branch", branch_id: ramo.id }) : null;
              })();
          if (!edge) {
            return {
              kind: "fail",
              error: `match_reply node "${node.id}" has no edge for branch "${hit?.id ?? "else"}" (fallback also missing)`,
            };
          }
          return { kind: "advance", next_node_id: edge.target, next_eval_at: clock() };
        }
      }
      const edge = selectEdge(edges, node.id, classEdgeMatch(node, NO_REPLY_BRANCH_ID));
      if (!edge) {
        return {
          kind: "fail",
          error: `match_reply node "${node.id}" has no edge for class "no_reply" (fallback also missing)`,
        };
      }
      return { kind: "advance", next_node_id: edge.target, next_eval_at: clock() };
    }

    case "repeat": {
      const taken = repeatTaken ?? 0;
      let total = repeatTotal ?? null;
      if (total === null) {
        const parsed = parseReplyCount(lastInboundBody, node.config.max_count);
        if (parsed === null) {
          const edge = selectEdge(edges, node.id, { type: "always" });
          if (!edge) {
            return { kind: "fail", error: `repeat node "${node.id}" has no fallback edge for an unreadable count` };
          }
          return { kind: "advance", next_node_id: edge.target, next_eval_at: clock() };
        }
        total = parsed;
      }
      if (taken >= total) {
        const edge = selectEdge(edges, node.id, { type: "branch", branch_id: REPEAT_DONE_BRANCH_ID });
        if (!edge) {
          return { kind: "fail", error: `repeat node "${node.id}" has no edge for branch "${REPEAT_DONE_BRANCH_ID}"` };
        }
        return { kind: "advance", next_node_id: edge.target, next_eval_at: clock(), repeat: { index: taken, total } };
      }
      const edge = selectEdge(edges, node.id, { type: "branch", branch_id: REPEAT_BODY_BRANCH_ID });
      if (!edge) {
        return { kind: "fail", error: `repeat node "${node.id}" has no edge for branch "${REPEAT_BODY_BRANCH_ID}"` };
      }
      return {
        kind: "advance",
        next_node_id: edge.target,
        next_eval_at: clock(),
        repeat: { index: taken + 1, total },
      };
    }

    case "collect": {
      // Nó de COLETA do fluxo de atendimento (surface=atendimento). Perguntar e
      // gravar é responsabilidade do executor in-turn; no relógio do follow-up
      // ele é passagem (segue pela aresta única). Um fluxo de retomada não
      // deveria usar este nó — o publish é quem recorta isso.
      const edge = selectEdge(edges, node.id, { type: "always" });
      if (!edge) return { kind: "fail", error: `collect node "${node.id}" has no outbound edge` };
      return { kind: "advance", next_node_id: edge.target, next_eval_at: clock() };
    }

    case "skill": {
      // Puxa uma skill instalada em paralelo ao passo; a ativação é do executor
      // in-turn (união com o `matchSkills`). No relógio, é passagem.
      const edge = selectEdge(edges, node.id, { type: "always" });
      if (!edge) return { kind: "fail", error: `skill node "${node.id}" has no outbound edge` };
      return { kind: "advance", next_node_id: edge.target, next_eval_at: clock() };
    }

    case "action": {
      // At-most-once send: enqueue the turn EXACTLY ONCE per occupancy. First entry
      // (no prior occupancy event) enqueues; a recheck fired while the turn is still in
      // flight — completeTurnForEnrollment (turn-bridge) hasn't advanced the enrollment
      // yet — must NOT re-enqueue. Mirrors the wait/ai_classify guard (resolveWaitPhase),
      // which the action node lacked (steps_taken increments every recheck, so the
      // `${node}:${steps}` idempotency_key was a FRESH key each tick → a 2nd job → a 2nd
      // real send that the send sink's (job_id,seq) dedup can't catch).
      if (!actionEnqueued && !actionCompleted) {
        if (
          proximo?.type === "match_reply" &&
          proximo.config.save_to &&
          destinoJaPreenchido(lead, proximo.config.save_to)
        ) {
          const modo = modoSeJaExiste(proximo);
          if (modo === "skip" || modo === "confirm") {
            const edge = selectEdge(edges, node.id, { type: "always" });
            if (!edge) return { kind: "fail", error: `action node "${node.id}" has no outbound edge` };
            return { kind: "advance", next_node_id: edge.target, next_eval_at: clock() };
          }
        }
        return { kind: "enqueue_turn", purpose: "send_message", wake_status: "active" };
      }
      // Envio já fechou (action_sent) mas o enrollment ainda está no action —
      // típico de corrida: completeTurn avançou e um recheck concorrente reverteu.
      if (actionCompleted) {
        const edge = selectEdge(edges, node.id, { type: "always" });
        if (!edge) return { kind: "fail", error: `action node "${node.id}" has no outbound edge` };
        return { kind: "advance", next_node_id: edge.target, next_eval_at: clock() };
      }
      // Dead-man: the turn never completed. Rechecks count THIS occupancy only
      // (`occupancyEventCount`) so a `repeat` that volta no mesmo nó de ação não
      // herda o orçamento das voltas anteriores.
      if ((actionRecheckCount ?? 0) >= MAX_ACTION_RECHECKS) {
        return { kind: "dead", reason: "action_turn_never_completed" };
      }
      return {
        kind: "recheck",
        next_eval_at: new Date(clock().getTime() + atrasoDoRecheck(actionRecheckCount ?? 0)),
      };
    }

    case "internal_task": {
      // Lembrete interno (#1540): este nó NÃO enfileira turno de envio — é a
      // diferença inteira da feature. Ele avança, e quem grava a tarefa é o
      // engine ao aplicar o `advance` (`criarTarefaInterna`), guardado pelo
      // MESMO idempotency_key do evento do passo: replay do tick não cria a
      // segunda tarefa, e um fluxo "somente interno" não tem mensagem nenhuma
      // para sair.
      const edge = selectEdge(edges, node.id, { type: "always" });
      if (!edge) return { kind: "fail", error: `internal_task node "${node.id}" has no outbound edge` };
      return { kind: "advance", next_node_id: edge.target, next_eval_at: clock() };
    }

    case "move_lead": {
      // #2065 — mover o card de etapa é PASSAGEM no relógio, como o
      // `internal_task`: o nó avança pela aresta única e quem ESCREVE a etapa é
      // o motor ao aplicar o `advance` (`db.moverLeadNoFunil`, que chama o
      // `moveLeadHandler` da casa com o `lost_reason` do bloco, sem padrão
      // escondido), guardado pelo idempotency_key do evento do
      // passo — replay do tick não move o card duas vezes.
      const edge = selectEdge(edges, node.id, { type: "always" });
      if (!edge) return { kind: "fail", error: `move_lead node "${node.id}" has no outbound edge` };
      return { kind: "advance", next_node_id: edge.target, next_eval_at: clock() };
    }

    case "edit_lead_tag": {
      // #2065 — mesma passagem do `move_lead`: a tag nasce no motor, depois do
      // evento do passo, com a MESMA trava. Nenhuma mensagem sai daqui (por isso
      // este nó não está em `NOS_QUE_ENVIAM`).
      const edge = selectEdge(edges, node.id, { type: "always" });
      if (!edge) return { kind: "fail", error: `edit_lead_tag node "${node.id}" has no outbound edge` };
      return { kind: "advance", next_node_id: edge.target, next_eval_at: clock() };
    }

    case "end": {
      if (node.config.outcome === "custom") {
        return { kind: "complete", outcome: null, cancel_reason: node.config.note };
      }
      return { kind: "complete", outcome: node.config.outcome };
    }
  }
}