import { afterAll, beforeEach, describe, expect, it } from "vitest";
import pg from "pg";

import {
  runSilenceSweep,
  type SilenceSweepDb,
  type SilencePointer,
} from "@/lib/followup/silence-sweep";
import { isPointerEnabledForAutomaticTrigger, noDeGatilhoDoGrafo, type FollowupGateDb } from "@/lib/followup/agent-followup-gate";
import { CONVERSATION_TERMINAL_STATUSES } from "@/lib/schemas";
import type { FlowGraph } from "@/lib/followup/graph-schema";

/**
 * Task 8.1 — gatilho de silêncio (varredura TIME-DRIVEN no cron) contra
 * Postgres real. DESKCOMM_GOV_INVARIANTS_EDIT=1 — arquivo NOVO desta sessão
 * (tests/invariants/** está congelado pro resto, não para arquivos próprios).
 *
 * Congela: (1) pointer silence habilitado no gate + contato silêncio >
 * threshold → exatamente 1 enrollment nascendo no nó trigger; rodar a
 * varredura DE NOVO não duplica (unique-live, `idx_followup_enrollments_one_live`);
 * (2) o MESMO cenário mas SEM nenhum agente: grafo só de texto enrolla com
 * `agent_id` nulo; grafo que pede IA é gate-out;
 * (3) contato silencioso HÁ MENOS que o threshold → não enrolla (boundary);
 * silêncio EXATAMENTE igual ao threshold → enrolla (`<=`, não `<`);
 * (4) `isPointerEnabledForAutomaticTrigger` contra `ai_agent_versions` REAL
 * (não o fake hand-rolled de agent-followup-gate.test.ts): publicado+
 * habilitado+pointer-membro → true; rascunho (não publicado) → false;
 * enabled=false → false; pointer de OUTRA org → false (reviewer Minor #2 da
 * Task 7.2 — a task que primeiro CONSOME o gate é quem prova a query real);
 * (5) redução anti-spam (review pós-approval, o gap coberto agora): contato
 * com 2 conversas (1 velha silenciosa + 1 recente dentro do threshold) NÃO
 * enrolla — prova que a redução pega o `last_inbound_at` MAIS RECENTE por
 * contato, não a 1ª conversa velha encontrada; contato cuja única conversa
 * nunca recebeu inbound (`last_inbound_at is null`) também NÃO enrolla —
 * "nunca conversou" ≠ "está em silêncio", sem mass-enroll de contato novo.
 *
 * (1b) cooldown pós-conclusão (issue de produção, 2026-09-25): um enrollment
 * deste MESMO pointer que terminou há menos que `threshold_minutes` barra
 * reinscrição, mesmo com o contato continuando silencioso (o cutoff de
 * silêncio sozinho reinscreveria a cada tick do cron); terminado há mais que
 * `threshold_minutes`, reinscreve normalmente.
 */

const container = process.env.TEST_DB_CONTAINER;
if (!container) {
  throw new Error("TEST_DB_CONTAINER not set — rode via `pnpm test:invariants` (scripts/test-db.sh)");
}

const PORT = Number(process.env.TEST_DB_PORT ?? 54329);
const pool = new pg.Pool({
  connectionString: `postgresql://postgres:postgres@127.0.0.1:${PORT}/postgres`,
  max: 4,
});

afterAll(async () => {
  await pool.end();
});

// `loadActiveSilencePointers` é GLOBAL (cross-org, mesmo desenho de
// `fn_claim_due_followup_enrollments`) — um pointer kind='silence' de um `it`
// anterior (status='active' pra sempre, nada o desativa) seria re-escaneado
// pelo `it` seguinte e contaminaria pointers_scanned/pointers_gated_out (mesma
// classe de flake já resolvida em followup-engine.test.ts, Task 5.2 fix 2).
// Escopo do delete: só kind='silence' — não toca pointers de OUTROS arquivos
// (reactivity/engine/turn-bridge nunca usam kind='silence'). Cascade:
// `followup_enrollments.pointer_id` tem ON DELETE CASCADE (migration 0054).
beforeEach(async () => {
  await pool.query(`delete from followup_flow_pointers where trigger_config->>'kind' = 'silence'`);
});

// ---- pg-backed SilenceSweepDb (test-only; prod usa createSupabaseSilenceSweepDb) ----

/**
 * O SEGUNDO IMPLEMENTADOR DE `SilenceSweepDb` — e o primeiro é
 * `createSupabaseSilenceSweepDb`, em `lib/followup/silence-sweep.ts`.
 *
 * A LÓGICA testada aqui é a real: este arquivo importa e chama
 * `runSilenceSweep`. O que é dublê é a BORDA de acesso a dados — a produção
 * fala com o PostgREST e o `test:db` sobe só o Postgres, então cada método
 * abaixo traduz a consulta de lá para SQL. Dois implementadores do mesmo
 * contrato, e eles têm de concordar.
 *
 * ⚠️ Eles JÁ divergiram, e o custo apareceu no PR #420 (@automatikpg-ux): a
 * produção ganhou `.not("status","in", …)` para não cobrar quem um humano já
 * encerrou, e esta cópia ficou sem o filtro. Os dois casos que aquele PR
 * escreveu para provar a intenção reprovaram medindo A CÓPIA, não o original.
 *
 * ═══ O que a constante compartilhada garante, e o que NÃO garante ═══
 *
 * O filtro abaixo é montado a partir de `CONVERSATION_TERMINAL_STATUSES`, a
 * mesma que a produção usa — e não de `'closed','archived'` escrito à mão. A
 * diferença não é cosmética: acrescentar um status terminal passa a mudar os
 * DOIS lados no mesmo commit, sem ninguém precisar lembrar deste arquivo.
 *
 * Isso prende o **valor**. Não prende a **forma**: um `.eq()` novo na produção
 * que este SQL não tenha continua invisível daqui. Esse eixo é dívida
 * declarada, com item de plano próprio — a pergunta dele é "como fazer os dois
 * adaptadores passarem pela mesma bateria de contrato", e não cabe num PR de
 * contribuidor. Escrever o escopo é o que impede esta cerca de anestesiar quem
 * vier depois achando que ela cobre tudo.
 */
function silenceSweepDb(): SilenceSweepDb {
  return {
    async loadActiveSilencePointers(): Promise<SilencePointer[]> {
      const { rows } = await pool.query<{
        id: string;
        organization_id: string;
        active_version_id: string | null;
        trigger_config: {
          kind: string;
          params?: {
            threshold_minutes: number;
            segments?: string[];
            reentry_pause_minutes?: number;
            max_silence_minutes?: number;
            reentry_pause_basis?: "ultima_mensagem" | "ultimo_envio";
          };
        };
        handoff_policy: "pause" | "cancel" | "allow";
      }>(
        `select id, organization_id, active_version_id, trigger_config, handoff_policy
         from followup_flow_pointers
         where status = 'active' and active_version_id is not null`,
      );
      const pointers: SilencePointer[] = [];
      for (const row of rows) {
        if (row.trigger_config.kind !== "silence" || !row.active_version_id) continue;
        pointers.push({
          id: row.id,
          organization_id: row.organization_id,
          active_version_id: row.active_version_id,
          threshold_minutes: row.trigger_config.params!.threshold_minutes,
          segments: row.trigger_config.params!.segments ?? [],
          reentry_pause_minutes: row.trigger_config.params!.reentry_pause_minutes ?? 0,
          ...(row.trigger_config.params!.max_silence_minutes
            ? { max_silence_minutes: row.trigger_config.params!.max_silence_minutes }
            : {}),
          ...(row.trigger_config.params!.reentry_pause_basis
            ? { reentry_pause_basis: row.trigger_config.params!.reentry_pause_basis }
            : {}),
          handoff_policy: row.handoff_policy,
        });
      }
      return pointers;
    },
    async loadSilentContactIds(orgId, cutoffIso, segments, desdeIso) {
      const { rows } = await pool.query<{ contact_id: string; last_inbound_at: string; tags: string[]; is_blocked: boolean }>(
        `select conv.contact_id, max(conv.last_inbound_at) as last_inbound_at,
                c.tags as tags, c.is_blocked as is_blocked
         from conversations conv
         join contacts c on c.id = conv.contact_id
         where conv.organization_id = $1 and conv.last_inbound_at is not null
           and conv.status <> all($2::text[])
         group by conv.contact_id, c.tags, c.is_blocked`,
        [orgId, CONVERSATION_TERMINAL_STATUSES],
      );
      const cutoff = new Date(cutoffIso).getTime();
      const desde = desdeIso ? new Date(desdeIso).getTime() : null;
      return rows
        .filter((r) => !r.is_blocked)
        .filter((r) => new Date(r.last_inbound_at).getTime() <= cutoff)
        .filter((r) => desde === null || new Date(r.last_inbound_at).getTime() >= desde)
        .filter((r) => segments.length === 0 || segments.some((s) => r.tags.includes(s)))
        .map((r) => r.contact_id);
    },
    // Mesma régua de `lib/followup/retorno-segura-o-fluxo.ts`: retorno agendado,
    // que não é o turno de um passo de fluxo.
    async loadContatosComRetornoVivo(orgId) {
      const { rows } = await pool.query<{ contact_id: string }>(
        `select contact_id from cron_jobs
          where organization_id = $1 and kind = 'at' and job_kind = 'followup_turn'
            and enabled and cancelled_at is null and contact_id is not null
            and payload->>'followup_enrollment_id' is null`,
        [orgId],
      );
      return new Set(rows.map((r) => r.contact_id));
    },
    // Mesma régua de `createSupabaseSilenceSweepDb`: inscrição ENCERRADA deste
    // fluxo (o mais tardio entre `completed_at` e `updated_at` é o fim) e a
    // última mensagem do contato.
    async loadEncerramentosDoFluxo(orgId, pointerId, contactIds) {
      const { rows } = await pool.query<{ contact_id: string; fim: Date; ultima: Date | null }>(
        `select e.contact_id,
                max(greatest(coalesce(e.completed_at, e.updated_at), e.updated_at)) as fim,
                (select max(conv.last_inbound_at) from conversations conv
                  where conv.organization_id = $1 and conv.contact_id = e.contact_id) as ultima
           from followup_enrollments e
          where e.organization_id = $1 and e.pointer_id = $2
            and e.status in ('completed', 'cancelled', 'dead')
            and e.contact_id = any($3::uuid[])
          group by e.contact_id`,
        [orgId, pointerId, contactIds],
      );
      return new Map(
        rows.map((r) => [r.contact_id, { encerradaEm: r.fim.getTime(), ultimaMensagemEm: r.ultima?.getTime() ?? null }]),
      );
    },
    // A conversa MAIS RECENTE do contato decide, como na produção.
    async loadContatosComPessoaNoComando(orgId, contactIds) {
      const { rows } = await pool.query<{ contact_id: string }>(
        `select contact_id from (
           select distinct on (conv.contact_id) conv.contact_id,
                  (conv.assignee_kind = 'user' or c.force_human
                   or coalesce(conv.bot_silenced_until > now(), false)) as com_pessoa
             from conversations conv
             join contacts c on c.id = conv.contact_id
            where conv.organization_id = $1 and conv.contact_id = any($2::uuid[])
            order by conv.contact_id, conv.last_inbound_at desc nulls last
         ) x where com_pessoa`,
        [orgId, contactIds],
      );
      return new Set(rows.map((r) => r.contact_id));
    },
    // Mesma pergunta do adaptador de produção: vivo em QUALQUER fluxo da org.
    async loadContatosComInscricaoViva(orgId, contactIds) {
      const { rows } = await pool.query<{ contact_id: string }>(
        `select distinct contact_id from followup_enrollments
          where organization_id = $1 and contact_id = any($2::uuid[])
            and status in ('active','waiting_reply','paused_handoff','paused_manual')`,
        [orgId, contactIds],
      );
      return new Set(rows.map((r) => r.contact_id));
    },
    async loadTriggerNode(orgId, versionId) {
      const { rows } = await pool.query<{ graph: FlowGraph }>(
        `select graph from followup_flow_versions where organization_id = $1 and id = $2`,
        [orgId, versionId],
      );
      if (rows.length === 0) return null;
      return noDeGatilhoDoGrafo(rows[0]!.graph);
    },
    async loadContactIdsEmCooldown(orgId, pointerId, contactIds, cutoffIso) {
      if (contactIds.length === 0) return new Set();
      const { rows } = await pool.query<{ contact_id: string }>(
        `select distinct contact_id from followup_enrollments
         where organization_id = $1 and pointer_id = $2 and contact_id = any($3::uuid[])
           and status not in ('active','waiting_reply','paused_handoff','paused_manual')
           and updated_at >= $4`,
        [orgId, pointerId, contactIds, cutoffIso],
      );
      return new Set(rows.map((r) => r.contact_id));
    },
    async insertEnrollment(input) {
      try {
        await pool.query(
          `insert into followup_enrollments
             (organization_id, pointer_id, version_id, contact_id, current_node_id, status, next_eval_at, agent_id)
           values ($1, $2, $3, $4, $5, 'active', $6, $7)`,
          [
            input.organization_id,
            input.pointer_id,
            input.version_id,
            input.contact_id,
            input.current_node_id,
            input.next_eval_at,
            input.agent_id,
          ],
        );
        return { inserted: true };
      } catch (err) {
        if ((err as { code?: string }).code === "23505") return { inserted: false };
        throw err;
      }
    },
  };
}

// ---- pg-backed FollowupGateDb (mirrors createSupabaseFollowupGateDb's query, against real ai_agent_versions) ----

function pgGateDb(): FollowupGateDb {
  return {
    async loadEnabledPublishedFollowupAgents(orgId) {
      const { rows } = await pool.query<{ agent_id: string; followup: unknown }>(
        `select agent_id, followup from ai_agent_versions where organization_id = $1 and status = 'published'`,
        [orgId],
      );
      const byAgent = new Map<string, Set<string>>();
      for (const row of rows) {
        const f = row.followup as { enabled?: unknown; flow_pointer_ids?: unknown } | null;
        if (!f || f.enabled !== true || !Array.isArray(f.flow_pointer_ids)) continue;
        const set = byAgent.get(row.agent_id) ?? new Set<string>();
        for (const id of f.flow_pointer_ids) if (typeof id === "string") set.add(id);
        if (set.size > 0) byAgent.set(row.agent_id, set);
      }
      return [...byAgent].map(([agentId, ids]) => ({ agentId, pointerIds: [...ids] }));
    },
  };
}

// ---- seed helpers ----

let orgSeq = 0;
function nextOrgId(): string {
  orgSeq += 1;
  return `dddddd${String(orgSeq).padStart(2, "0")}-0000-4000-8000-000000000001`;
}

async function seedOrg(org: string): Promise<void> {
  const name = `followup-silence-${org.slice(0, 8)}`;
  await pool.query(
    `insert into organizations (id, slug, legal_name, display_name) values ($1, $2, $3, $4) on conflict (id) do nothing`,
    [org, name, name, name],
  );
}

async function seedContact(
  org: string,
  opts?: { tags?: string[]; isBlocked?: boolean; forceHuman?: boolean },
): Promise<string> {
  const { rows } = await pool.query<{ id: string }>(
    `insert into contacts (organization_id, display_name, tags, is_blocked, force_human)
     values ($1, 'Silence Contact', $2, $3, $4) returning id`,
    [org, opts?.tags ?? [], opts?.isBlocked ?? false, opts?.forceHuman ?? false],
  );
  return rows[0]!.id;
}

async function seedChannelSession(org: string): Promise<string> {
  const { rows } = await pool.query<{ id: string }>(
    `insert into channel_sessions (organization_id, waha_session_name, status, webhook_secret_encrypted)
     values ($1, $2, 'WORKING', '\\x00'::bytea) returning id`,
    [org, `silence-session-${Date.now()}-${Math.random()}`],
  );
  return rows[0]!.id;
}

/** `agoMinutes=null` → conversa sem last_inbound_at (nunca recebeu inbound). */
async function seedConversation(org: string, contactId: string, agoMinutes: number | null): Promise<string> {
  const sessionId = await seedChannelSession(org);
  const lastInboundExpr = agoMinutes === null ? "null" : `now() - interval '${agoMinutes} minutes'`;
  const { rows } = await pool.query<{ id: string }>(
    `insert into conversations (organization_id, contact_id, channel_session_id, status, is_group, last_inbound_at)
     values ($1, $2, $3, 'open', false, ${lastInboundExpr}) returning id`,
    [org, contactId, sessionId],
  );
  return rows[0]!.id;
}

/** Mesmo que `seedConversation`, mas com `status` explícito — pra provar que uma conversa
 *  CLOSED/ARCHIVED não conta como silêncio (um humano encerrou; o fluxo não está mais ativo). */
async function seedConversationComStatus(
  org: string,
  contactId: string,
  agoMinutes: number,
  status: string,
): Promise<string> {
  const sessionId = await seedChannelSession(org);
  const { rows } = await pool.query<{ id: string }>(
    `insert into conversations (organization_id, contact_id, channel_session_id, status, is_group, last_inbound_at)
     values ($1, $2, $3, $4, false, now() - interval '${agoMinutes} minutes') returning id`,
    [org, contactId, sessionId, status],
  );
  return rows[0]!.id;
}

/** Mesmo que `seedConversation`, mas com `last_inbound_at` EXATO (ISO), não relativo a `now()`
 *  do Postgres — necessário pro teste de boundary exato (evita drift entre `now()` do DB e o
 *  `clock()` injetado no sweep, que roda em JS). */
async function seedConversationAt(org: string, contactId: string, atIso: string): Promise<string> {
  const sessionId = await seedChannelSession(org);
  const { rows } = await pool.query<{ id: string }>(
    `insert into conversations (organization_id, contact_id, channel_session_id, status, is_group, last_inbound_at)
     values ($1, $2, $3, 'open', false, $4) returning id`,
    [org, contactId, sessionId, atIso],
  );
  return rows[0]!.id;
}

async function seedSilenceFlow(
  org: string,
  opts?: {
    thresholdMinutes?: number;
    segments?: string[];
    comIa?: boolean;
    reentryPauseMinutes?: number;
    handoffPolicy?: "pause" | "cancel" | "allow";
    maxSilenceMinutes?: number;
    reentryPauseBasis?: "ultima_mensagem" | "ultimo_envio";
  },
): Promise<{ pointerId: string; versionId: string }> {
  const graph = {
    nodes: [
      { id: "t1", type: "trigger", label: "Start", position: { x: 0, y: 0 }, config: {} },
      ...(opts?.comIa
        ? [{ id: "c1", type: "ai_classify" as const, label: "Class", position: { x: 0, y: 0 }, config: {} }]
        : []),
      { id: "e1", type: "end", label: "Done", position: { x: 0, y: 0 }, config: { outcome: "converted" } },
    ],
    edges: [{ id: "t1-e1", source: "t1", target: "e1", priority: 0, condition: { type: "always" } }],
  } as FlowGraph;
  const { rows: versionRows } = await pool.query<{ id: string }>(
    `insert into followup_flow_versions (organization_id, graph) values ($1, $2) returning id`,
    [org, JSON.stringify(graph)],
  );
  const versionId = versionRows[0]!.id;
  const triggerConfig = {
    kind: "silence",
    params: {
      threshold_minutes: opts?.thresholdMinutes ?? 60,
      segments: opts?.segments ?? [],
      ...(opts?.reentryPauseMinutes !== undefined ? { reentry_pause_minutes: opts.reentryPauseMinutes } : {}),
      ...(opts?.maxSilenceMinutes !== undefined ? { max_silence_minutes: opts.maxSilenceMinutes } : {}),
      ...(opts?.reentryPauseBasis !== undefined ? { reentry_pause_basis: opts.reentryPauseBasis } : {}),
    },
  };
  const { rows: pointerRows } = await pool.query<{ id: string }>(
    `insert into followup_flow_pointers (organization_id, name, status, active_version_id, trigger_config, handoff_policy)
     values ($1, $2, 'active', $3, $4, $5) returning id`,
    [
      org,
      `Silence Flow ${Date.now()}-${Math.random()}`,
      versionId,
      JSON.stringify(triggerConfig),
      opts?.handoffPolicy ?? "pause",
    ],
  );
  return { pointerId: pointerRows[0]!.id, versionId };
}

async function seedPublishedAgentVersion(
  org: string,
  opts: { status?: string; enabled?: boolean; pointerIds?: string[]; agentId?: string },
): Promise<string> {
  const sessionId = await seedChannelSession(org);
  const { rows: agentRows } = await pool.query<{ id: string }>(
    opts.agentId
      ? `insert into ai_agents (id, organization_id, name, system_prompt) values ($3, $1, $2, 'prompt') returning id`
      : `insert into ai_agents (organization_id, name, system_prompt) values ($1, $2, 'prompt') returning id`,
    opts.agentId
      ? [org, `Silence Gate Agent ${Date.now()}-${Math.random()}`, opts.agentId]
      : [org, `Silence Gate Agent ${Date.now()}-${Math.random()}`],
  );
  const agentId = agentRows[0]!.id;
  const followup = { enabled: opts.enabled ?? true, flow_pointer_ids: opts.pointerIds ?? [] };
  await pool.query(
    `insert into ai_agent_versions
       (organization_id, agent_id, version_number, system_prompt, provider, model, channel_session_id, status, followup)
     values ($1, $2, 1, 'prompt', 'anthropic', 'claude-sonnet-4-6', $3, $4, $5)`,
    [org, agentId, sessionId, opts.status ?? "published", JSON.stringify(followup)],
  );
  return agentId;
}

async function countEnrollments(pointerId: string, contactId: string): Promise<number> {
  const { rows } = await pool.query<{ n: string }>(
    `select count(*) as n from followup_enrollments where pointer_id = $1 and contact_id = $2`,
    [pointerId, contactId],
  );
  return Number(rows[0]!.n);
}

const CLOCK = () => new Date();

// ---- 1. sweep enrolla + idempotência ------------------------------------

describe("runSilenceSweep — enrolla contato silencioso gateado, sem duplicar", () => {
  it("pointer silence habilitado + contato silêncio > threshold → 1 enrollment; 2ª varredura não duplica", async () => {
    const org = nextOrgId();
    await seedOrg(org);
    const { pointerId, versionId } = await seedSilenceFlow(org, { thresholdMinutes: 30 });
    await seedPublishedAgentVersion(org, { enabled: true, pointerIds: [pointerId] });
    const contactId = await seedContact(org);
    await seedConversation(org, contactId, 90); // silencioso há 90min > threshold 30min

    const deps = { db: silenceSweepDb(), gateDb: pgGateDb(), clock: CLOCK };

    const summary1 = await runSilenceSweep(deps);
    expect(summary1.pointers_scanned).toBeGreaterThanOrEqual(1);
    expect(summary1.enrolled).toBe(1);
    expect(summary1.skipped_existing).toBe(0);

    const enrollment = await pool.query<{ current_node_id: string; status: string }>(
      `select current_node_id, status from followup_enrollments where pointer_id = $1 and contact_id = $2`,
      [pointerId, contactId],
    );
    expect(enrollment.rows).toHaveLength(1);
    expect(enrollment.rows[0]!.current_node_id).toBe("t1");
    expect(enrollment.rows[0]!.status).toBe("active");

    // 2ª varredura: unique-live index barra duplicata — vira skipped_existing, não erro.
    const summary2 = await runSilenceSweep(deps);
    expect(summary2.enrolled).toBe(0);
    expect(summary2.skipped_existing).toBeGreaterThanOrEqual(1);
    expect(await countEnrollments(pointerId, contactId)).toBe(1);

    expect(versionId).toBeTruthy(); // sanity — version foi realmente usada (current_node_id veio do grafo pinado nela)
  });
});

// ---- 1b. cooldown pós-conclusão (issue de produção, 2026-09-25) --------

describe("runSilenceSweep — cooldown pós-conclusão (não reinscreve a cada tick do cron)", () => {
  it("enrollment deste pointer TERMINOU há menos que threshold_minutes → NÃO reinscreve (contato continua silencioso)", async () => {
    const org = nextOrgId();
    await seedOrg(org);
    const thresholdMinutes = 120;
    const { pointerId, versionId } = await seedSilenceFlow(org, { thresholdMinutes });
    const agentId = await seedPublishedAgentVersion(org, { enabled: true, pointerIds: [pointerId] });
    const contactId = await seedContact(org);
    await seedConversation(org, contactId, 200); // bem além do threshold — o gatilho de silêncio sozinho enrollaria

    // Simula o enrollment que a varredura ANTERIOR já criou e concluiu — a
    // mensagem de nudge já saiu há 3 minutos, não há 2 horas. `updated_at`
    // gravado igual a `completed_at`, como o motor real faz (engine.ts:301-302).
    await pool.query(
      `insert into followup_enrollments
         (organization_id, pointer_id, version_id, contact_id, current_node_id, status,
          next_eval_at, agent_id, started_at, completed_at, updated_at, outcome)
       values ($1,$2,$3,$4,'e1','completed', null, $5,
               now() - interval '3 minutes', now() - interval '1 minute', now() - interval '1 minute',
               'exhausted')`,
      [org, pointerId, versionId, contactId, agentId],
    );

    const summary = await runSilenceSweep({ db: silenceSweepDb(), gateDb: pgGateDb(), clock: CLOCK });
    expect(summary.enrolled).toBe(0);
    expect(summary.skipped_cooldown).toBeGreaterThanOrEqual(1);
    expect(await countEnrollments(pointerId, contactId)).toBe(1); // continua só o antigo — nenhum novo nasceu
  });

  it("mesmo cenário, mas a última tentativa TERMINOU há mais que threshold_minutes → reinscreve normalmente", async () => {
    const org = nextOrgId();
    await seedOrg(org);
    const thresholdMinutes = 30;
    const { pointerId, versionId } = await seedSilenceFlow(org, { thresholdMinutes });
    const agentId = await seedPublishedAgentVersion(org, { enabled: true, pointerIds: [pointerId] });
    const contactId = await seedContact(org);
    await seedConversation(org, contactId, 90);

    await pool.query(
      `insert into followup_enrollments
         (organization_id, pointer_id, version_id, contact_id, current_node_id, status,
          next_eval_at, agent_id, started_at, completed_at, updated_at, outcome)
       values ($1,$2,$3,$4,'e1','completed', null, $5,
               now() - interval '40 minutes', now() - interval '38 minutes', now() - interval '38 minutes',
               'exhausted')`,
      [org, pointerId, versionId, contactId, agentId],
    );

    const summary = await runSilenceSweep({ db: silenceSweepDb(), gateDb: pgGateDb(), clock: CLOCK });
    expect(summary.skipped_cooldown).toBe(0);
    expect(summary.enrolled).toBe(1);
    expect(await countEnrollments(pointerId, contactId)).toBe(2); // o antigo + o novo
  });

  it("fluxo LENTO (started_at fora da janela, mas TERMINOU agora) → ainda em cooldown pelo fim, não pelo início", async () => {
    // Reproduz o achado da revisão: um nó `wait` do grafo aceita até 90 dias
    // (graph-schema.ts). Se o cooldown olhasse `started_at`, este caso voltaria
    // a reinscrever na hora — o mesmo defeito que esta PR existe para consertar.
    const org = nextOrgId();
    await seedOrg(org);
    const thresholdMinutes = 120;
    const { pointerId, versionId } = await seedSilenceFlow(org, { thresholdMinutes });
    const agentId = await seedPublishedAgentVersion(org, { enabled: true, pointerIds: [pointerId] });
    const contactId = await seedContact(org);
    await seedConversation(org, contactId, 200);

    await pool.query(
      `insert into followup_enrollments
         (organization_id, pointer_id, version_id, contact_id, current_node_id, status,
          next_eval_at, agent_id, started_at, completed_at, updated_at, outcome)
       values ($1,$2,$3,$4,'e1','completed', null, $5,
               now() - interval '5 days', now() - interval '1 minute', now() - interval '1 minute',
               'exhausted')`,
      [org, pointerId, versionId, contactId, agentId],
    );

    const summary = await runSilenceSweep({ db: silenceSweepDb(), gateDb: pgGateDb(), clock: CLOCK });
    expect(summary.enrolled).toBe(0);
    expect(summary.skipped_cooldown).toBeGreaterThanOrEqual(1);
    expect(await countEnrollments(pointerId, contactId)).toBe(1);
  });

  it("enrollment AINDA VIVO (status active) não conta como cooldown — segue sendo skipped_existing pelo índice único", async () => {
    // Se o cooldown enxergasse status vivos, este contato contaria como
    // `skipped_cooldown` em vez de `skipped_existing` — trocando o que os dois
    // contadores medem sem nenhuma mudança de comportamento real.
    const org = nextOrgId();
    await seedOrg(org);
    const { pointerId, versionId } = await seedSilenceFlow(org, { thresholdMinutes: 30 });
    const agentId = await seedPublishedAgentVersion(org, { enabled: true, pointerIds: [pointerId] });
    const contactId = await seedContact(org);
    await seedConversation(org, contactId, 90);

    await pool.query(
      `insert into followup_enrollments
         (organization_id, pointer_id, version_id, contact_id, current_node_id, status, next_eval_at, agent_id)
       values ($1,$2,$3,$4,'e1','active', now(), $5)`,
      [org, pointerId, versionId, contactId, agentId],
    );

    const summary = await runSilenceSweep({ db: silenceSweepDb(), gateDb: pgGateDb(), clock: CLOCK });
    expect(summary.skipped_cooldown).toBe(0);
    expect(summary.skipped_existing).toBeGreaterThanOrEqual(1);
    expect(await countEnrollments(pointerId, contactId)).toBe(1); // nenhum enrollment novo nasceu
  });
});

// ---- 2. gate: texto fixo sem agente enrolla; grafo de IA sem agente não -----

describe("runSilenceSweep — gate do agente", () => {
  it("contato silencioso, SEM agente, grafo só de texto → enrolla com agent_id nulo", async () => {
    const org = nextOrgId();
    await seedOrg(org);
    const { pointerId } = await seedSilenceFlow(org, { thresholdMinutes: 30 });
    const contactId = await seedContact(org);
    await seedConversation(org, contactId, 90);

    const summary = await runSilenceSweep({ db: silenceSweepDb(), gateDb: pgGateDb(), clock: CLOCK });
    expect(summary.pointers_gated_out).toBe(0);
    expect(summary.enrolled).toBeGreaterThanOrEqual(1);
    expect(await countEnrollments(pointerId, contactId)).toBe(1);
    const { rows } = await pool.query<{ agent_id: string | null }>(
      `select agent_id from followup_enrollments where pointer_id = $1 and contact_id = $2`,
      [pointerId, contactId],
    );
    expect(rows[0]!.agent_id).toBeNull();
  });

  it("contato silencioso, SEM agente, grafo que pede IA → 0 enrollments", async () => {
    const org = nextOrgId();
    await seedOrg(org);
    const { pointerId } = await seedSilenceFlow(org, { thresholdMinutes: 30, comIa: true });
    const contactId = await seedContact(org);
    await seedConversation(org, contactId, 90);

    const summary = await runSilenceSweep({ db: silenceSweepDb(), gateDb: pgGateDb(), clock: CLOCK });
    expect(summary.pointers_gated_out).toBeGreaterThanOrEqual(1);
    expect(await countEnrollments(pointerId, contactId)).toBe(0);
  });

  it("agente existe mas com followup.enabled=false, grafo de IA → gate-out também", async () => {
    const org = nextOrgId();
    await seedOrg(org);
    const { pointerId } = await seedSilenceFlow(org, { thresholdMinutes: 30, comIa: true });
    await seedPublishedAgentVersion(org, { enabled: false, pointerIds: [pointerId] });
    const contactId = await seedContact(org);
    await seedConversation(org, contactId, 90);

    const summary = await runSilenceSweep({ db: silenceSweepDb(), gateDb: pgGateDb(), clock: CLOCK });
    expect(summary.pointers_gated_out).toBeGreaterThanOrEqual(1);
    expect(await countEnrollments(pointerId, contactId)).toBe(0);
  });
});

// ---- 3. boundary: silêncio < threshold não enrolla ----------------------

describe("runSilenceSweep — boundary de threshold", () => {
  it("contato silencioso há MENOS que threshold_minutes → não enrolla", async () => {
    const org = nextOrgId();
    await seedOrg(org);
    const { pointerId } = await seedSilenceFlow(org, { thresholdMinutes: 60 });
    await seedPublishedAgentVersion(org, { enabled: true, pointerIds: [pointerId] });
    const contactId = await seedContact(org);
    await seedConversation(org, contactId, 10); // só 10min de silêncio, threshold=60

    const summary = await runSilenceSweep({ db: silenceSweepDb(), gateDb: pgGateDb(), clock: CLOCK });
    expect(summary.pointers_gated_out).toBe(0); // o gate passou — não é isso que bloqueou
    expect(summary.enrolled).toBe(0);
    expect(await countEnrollments(pointerId, contactId)).toBe(0);
  });

  it("silêncio EXATAMENTE igual ao threshold (at === cutoff) → enrolla (silêncio é <=, não <)", async () => {
    const org = nextOrgId();
    await seedOrg(org);
    const thresholdMinutes = 60;
    const { pointerId } = await seedSilenceFlow(org, { thresholdMinutes });
    await seedPublishedAgentVersion(org, { enabled: true, pointerIds: [pointerId] });
    const contactId = await seedContact(org);

    // clock() fixo (não `new Date()` a cada chamada) pra `cutoff` do sweep e o
    // `last_inbound_at` semeado baterem no MESMO instante, sem drift de execução.
    const refTime = new Date();
    const cutoff = new Date(refTime.getTime() - thresholdMinutes * 60_000);
    await seedConversationAt(org, contactId, cutoff.toISOString());

    const summary = await runSilenceSweep({ db: silenceSweepDb(), gateDb: pgGateDb(), clock: () => refTime });
    expect(summary.enrolled).toBe(1);
    expect(await countEnrollments(pointerId, contactId)).toBe(1);
  });
});

// ---- 3b. redução anti-spam (o ponto mais arriscado da lógica) -----------

describe("runSilenceSweep — redução anti-spam (multi-conversa + never-inbound)", () => {
  it("contato com 2 conversas — 1 antiga silenciosa + 1 RECENTE (dentro do threshold) → NÃO enrolla", async () => {
    const org = nextOrgId();
    await seedOrg(org);
    const { pointerId } = await seedSilenceFlow(org, { thresholdMinutes: 60 });
    await seedPublishedAgentVersion(org, { enabled: true, pointerIds: [pointerId] });
    const contactId = await seedContact(org);
    // conversa OLD: 120min de silêncio (> threshold 60, sozinha enrollaria).
    await seedConversation(org, contactId, 120);
    // conversa NEW: 5min — o contato respondeu recentemente por OUTRA conversa.
    await seedConversation(org, contactId, 5);

    const summary = await runSilenceSweep({ db: silenceSweepDb(), gateDb: pgGateDb(), clock: CLOCK });
    expect(summary.pointers_gated_out).toBe(0); // não foi o gate que bloqueou
    expect(summary.enrolled).toBe(0); // a redução MAX(last_inbound_at) pegou a conversa recente, não a velha
    expect(await countEnrollments(pointerId, contactId)).toBe(0);
  });

  it("única conversa está CLOSED (humano encerrou) e silenciosa há mais que o threshold → NÃO enrolla", async () => {
    const org = nextOrgId();
    await seedOrg(org);
    const { pointerId } = await seedSilenceFlow(org, { thresholdMinutes: 60 });
    await seedPublishedAgentVersion(org, { enabled: true, pointerIds: [pointerId] });
    const contactId = await seedContact(org);
    await seedConversationComStatus(org, contactId, 120, "closed");

    const summary = await runSilenceSweep({ db: silenceSweepDb(), gateDb: pgGateDb(), clock: CLOCK });
    expect(summary.enrolled).toBe(0);
    expect(await countEnrollments(pointerId, contactId)).toBe(0);
  });

  it("única conversa está ARCHIVED e silenciosa há mais que o threshold → NÃO enrolla", async () => {
    const org = nextOrgId();
    await seedOrg(org);
    const { pointerId } = await seedSilenceFlow(org, { thresholdMinutes: 60 });
    await seedPublishedAgentVersion(org, { enabled: true, pointerIds: [pointerId] });
    const contactId = await seedContact(org);
    await seedConversationComStatus(org, contactId, 120, "archived");

    const summary = await runSilenceSweep({ db: silenceSweepDb(), gateDb: pgGateDb(), clock: CLOCK });
    expect(summary.enrolled).toBe(0);
    expect(await countEnrollments(pointerId, contactId)).toBe(0);
  });

  it("conversa CLOSED antiga + conversa OPEN silenciosa (> threshold) do mesmo contato → enrolla pela ABERTA", async () => {
    const org = nextOrgId();
    await seedOrg(org);
    const { pointerId } = await seedSilenceFlow(org, { thresholdMinutes: 60 });
    await seedPublishedAgentVersion(org, { enabled: true, pointerIds: [pointerId] });
    const contactId = await seedContact(org);
    await seedConversationComStatus(org, contactId, 200, "closed"); // ignorada — não é a que decide
    await seedConversation(org, contactId, 90); // aberta, silenciosa > threshold

    const summary = await runSilenceSweep({ db: silenceSweepDb(), gateDb: pgGateDb(), clock: CLOCK });
    expect(summary.enrolled).toBe(1);
    expect(await countEnrollments(pointerId, contactId)).toBe(1);
  });

  it("contato cuja ÚNICA conversa nunca recebeu inbound (last_inbound_at NULL) → NÃO enrolla", async () => {
    const org = nextOrgId();
    await seedOrg(org);
    const { pointerId } = await seedSilenceFlow(org, { thresholdMinutes: 60 });
    await seedPublishedAgentVersion(org, { enabled: true, pointerIds: [pointerId] });
    const contactId = await seedContact(org);
    await seedConversation(org, contactId, null); // nunca recebeu inbound — não é "silêncio", é ausência de histórico

    const summary = await runSilenceSweep({ db: silenceSweepDb(), gateDb: pgGateDb(), clock: CLOCK });
    expect(summary.enrolled).toBe(0);
    expect(await countEnrollments(pointerId, contactId)).toBe(0);
  });
});

// ---- 4. gate SQL integration (contra ai_agent_versions REAL) ------------

describe("isPointerEnabledForAutomaticTrigger — integração SQL real (ai_agent_versions)", () => {
  it("publicado + enabled=true + pointer no array → true", async () => {
    const org = nextOrgId();
    await seedOrg(org);
    const { pointerId } = await seedSilenceFlow(org);
    await seedPublishedAgentVersion(org, { status: "published", enabled: true, pointerIds: [pointerId] });

    await expect(isPointerEnabledForAutomaticTrigger(pgGateDb(), org, pointerId)).resolves.toBe(true);
  });

  it("versão em rascunho (status='draft', não publicada) → false mesmo com enabled=true", async () => {
    const org = nextOrgId();
    await seedOrg(org);
    const { pointerId } = await seedSilenceFlow(org);
    await seedPublishedAgentVersion(org, { status: "draft", enabled: true, pointerIds: [pointerId] });

    await expect(isPointerEnabledForAutomaticTrigger(pgGateDb(), org, pointerId)).resolves.toBe(false);
  });

  it("publicado mas followup.enabled=false → false", async () => {
    const org = nextOrgId();
    await seedOrg(org);
    const { pointerId } = await seedSilenceFlow(org);
    await seedPublishedAgentVersion(org, { status: "published", enabled: false, pointerIds: [pointerId] });

    await expect(isPointerEnabledForAutomaticTrigger(pgGateDb(), org, pointerId)).resolves.toBe(false);
  });

  it("pointer habilitado numa org NÃO vaza pra outra org (cross-org)", async () => {
    const orgA = nextOrgId();
    const orgB = nextOrgId();
    await seedOrg(orgA);
    await seedOrg(orgB);
    const { pointerId } = await seedSilenceFlow(orgA);
    await seedPublishedAgentVersion(orgA, { status: "published", enabled: true, pointerIds: [pointerId] });

    await expect(isPointerEnabledForAutomaticTrigger(pgGateDb(), orgB, pointerId)).resolves.toBe(false);
  });
});

// ---- 6. Task 8.6: agent_id pinado no enrollment ------------------------

async function agentIdOfEnrollment(pointerId: string, contactId: string): Promise<string | null> {
  const { rows } = await pool.query<{ agent_id: string | null }>(
    `select agent_id from followup_enrollments where pointer_id = $1 and contact_id = $2`,
    [pointerId, contactId],
  );
  return rows[0]?.agent_id ?? null;
}

describe("runSilenceSweep — pina o agent_id do agente que arma o pointer (Task 8.6)", () => {
  it("um agente armando → agent_id do enrollment = esse agente", async () => {
    const org = nextOrgId();
    await seedOrg(org);
    const { pointerId } = await seedSilenceFlow(org, { thresholdMinutes: 30 });
    const agentId = await seedPublishedAgentVersion(org, { enabled: true, pointerIds: [pointerId] });
    const contactId = await seedContact(org);
    await seedConversation(org, contactId, 90);

    const summary = await runSilenceSweep({ db: silenceSweepDb(), gateDb: pgGateDb(), clock: CLOCK });
    expect(summary.enrolled).toBe(1);
    expect(await agentIdOfEnrollment(pointerId, contactId)).toBe(agentId);
  });

  it(">1 agente armando o MESMO pointer → pina o MENOR agent_id (uuid asc, determinístico)", async () => {
    const org = nextOrgId();
    await seedOrg(org);
    const { pointerId } = await seedSilenceFlow(org, { thresholdMinutes: 30 });
    const AGENT_LOW = "a0000000-0000-4000-8000-000000000001";
    const AGENT_HIGH = "f0000000-0000-4000-8000-000000000001";
    // ordem de seed HIGH antes de LOW — o pick não pode depender da ordem de inserção
    await seedPublishedAgentVersion(org, { enabled: true, pointerIds: [pointerId], agentId: AGENT_HIGH });
    await seedPublishedAgentVersion(org, { enabled: true, pointerIds: [pointerId], agentId: AGENT_LOW });
    const contactId = await seedContact(org);
    await seedConversation(org, contactId, 90);

    await runSilenceSweep({ db: silenceSweepDb(), gateDb: pgGateDb(), clock: CLOCK });
    expect(await agentIdOfEnrollment(pointerId, contactId)).toBe(AGENT_LOW);
  });
});

// ---- 7. Task 8.6: exclusividade 1-follow-up-vivo-por-lead ORG-WIDE ------

/**
 * Os status que ocupam a vaga do VIVO — copiados do baseline, nunca digitados
 * de memória. Três lugares deste arquivo dependem deles, e cada um por um
 * motivo diferente; o cabeçalho de `setOneLiveIndex` conta a história.
 */
const ONE_LIVE_STATUSES = "'active','waiting_reply','paused_handoff','paused_manual'";

/**
 * Quantos follow-ups VIVOS este contato tem.
 *
 * ⚠️ TERCEIRA CATEGORIA, e ela não é nenhuma das outras duas deste arquivo:
 * aqui não se restaura estado nem se reproduz história — faz-se uma PERGUNTA
 * sobre o estado atual. Por isso o predicado tem de ser o do produto HOJE, e
 * `paused_manual` entra: um enrollment pausado por uma pessoa ocupa a vaga do
 * vivo (é o que o índice garante desde a 0145), então uma contagem que o ignora
 * responde MENOS do que a realidade.
 *
 * Achado por `@QAVivo` na varredura do arquivo, e classificado por ele. Estava
 * LATENTE — medido: nenhuma asserção de hoje muda com a correção (16 verdes
 * antes, 16 depois). Consertado mesmo assim porque a armadilha estava armada: o
 * primeiro caso de pausa manual escrito aqui passaria por SORTE, que é
 * exatamente o defeito que esta wave já pagou uma vez.
 */
async function countLiveForContact(org: string, contactId: string): Promise<number> {
  const { rows } = await pool.query<{ n: string }>(
    `select count(*) as n from followup_enrollments
     where organization_id = $1 and contact_id = $2
       and status in (${ONE_LIVE_STATUSES})`,
    [org, contactId],
  );
  return Number(rows[0]!.n);
}

const ONE_LIVE = "idx_followup_enrollments_one_live";

/**
 * ⚠️ O PREDICADO É COPIADO DO BASELINE, NÃO DIGITADO DE MEMÓRIA.
 *
 * Este helper dropa e recria um índice do schema no banco COMPARTILHADO pelos
 * invariantes — e o `finally` que "restaura o estado do baseline" restaurava uma
 * versão CONGELADA no tempo: `('active','waiting_reply','paused_handoff')`, de
 * antes de a 0145 acrescentar `paused_manual`. Depois que este arquivo rodava, o
 * banco ficava com um índice que o baseline não tem mais, e o próximo teste a
 * depender dele reprovava — determinística e rapidamente (27ms, falha de
 * asserção, não timeout), o que fez a suíte parecer instável por carga.
 *
 * Medido: rodando só `followup-silence-sweep` + `followup-intervencao`, o caso
 * "pausado ocupa a vaga do contato na ORGANIZAÇÃO" reprova; isolado, passa. É o
 * `IA360-FLAKY` com assinatura. Os papéis, porque a diferença ensina: `@QAVivo`
 * CARACTERIZOU o fenômeno (determinístico, 36ms, não timeout) e REFUTOU a
 * primeira atribuição — a de que o índice da 0145 explicava tudo, que não
 * sobrevivia ao "isolado passa". Refutar uma causa errada é barato e uma medição
 * basta; achar a certa custou seguir a pista dele ("o culpado está na
 * vizinhança") até o par de arquivos.
 *
 * A lição que a constante carrega: quem recria objeto de schema num banco
 * compartilhado assume a dívida de acompanhar TODA migration futura que o toque.
 */
async function setOneLiveIndex(columns: string): Promise<void> {
  await pool.query(`drop index if exists ${ONE_LIVE}`);
  await pool.query(
    `create unique index if not exists ${ONE_LIVE} on followup_enrollments (${columns})
       where status in (${ONE_LIVE_STATUSES})`,
  );
}

describe("runSilenceSweep — 1 follow-up vivo por lead ORG-WIDE (Task 8.6, furo do Rafael)", () => {
  it("RED→GREEN: contato silencioso casa 2 fluxos — índice ANTIGO (pointer,contact) enrolla nos DOIS (spam); o NOVO (org,contact) barra o 2º", async () => {
    const org = nextOrgId();
    await seedOrg(org);
    const flowA = await seedSilenceFlow(org, { thresholdMinutes: 30 });
    const flowB = await seedSilenceFlow(org, { thresholdMinutes: 30 });
    // um agente publicado arma AMBOS os pointers
    await seedPublishedAgentVersion(org, { enabled: true, pointerIds: [flowA.pointerId, flowB.pointerId] });
    const contactId = await seedContact(org);
    await seedConversation(org, contactId, 90); // silencioso p/ os dois

    // Este caso mede o ÍNDICE: o pré-filtro de inscrição viva da varredura fica
    // desligado aqui, senão ele mesmo pularia o 2º fluxo e o RED (o spam do
    // índice antigo) nunca apareceria. O índice é a garantia; o pré-filtro é
    // economia — e tem caso próprio logo abaixo.
    const deps = {
      db: { ...silenceSweepDb(), loadContatosComInscricaoViva: async () => new Set<string>() },
      gateDb: pgGateDb(),
      clock: CLOCK,
    };

    try {
      // RED — índice como era ANTES da 0062: (pointer_id, contact_id)
      await setOneLiveIndex("pointer_id, contact_id");
      const red = await runSilenceSweep(deps);
      expect(red.enrolled).toBe(2); // um por fluxo — o spam que o Rafael achou
      expect(await countLiveForContact(org, contactId)).toBe(2);

      // limpa (2 vivos violariam o índice único org-wide ao recriá-lo)
      await pool.query(`delete from followup_enrollments where contact_id = $1`, [contactId]);

      // GREEN — índice da 0062: (organization_id, contact_id)
      await setOneLiveIndex("organization_id, contact_id");
      const green = await runSilenceSweep(deps);
      expect(green.enrolled).toBe(1); // exclusividade: só o 1º fluxo enrolla
      expect(green.skipped_existing).toBeGreaterThanOrEqual(1); // o 2º barrado por 23505 org-wide
      expect(await countLiveForContact(org, contactId)).toBe(1);
    } finally {
      // restaura o estado do baseline (índice org-wide), sem deixar linha viva órfã
      await pool.query(`drop index if exists ${ONE_LIVE}`);
      await pool.query(`delete from followup_enrollments where contact_id = $1`, [contactId]);
      await setOneLiveIndex("organization_id, contact_id");
    }
  });

  it("com o pré-filtro, o 2º fluxo nem TENTA inscrever quem já está vivo (sem 23505 a cada minuto)", async () => {
    const org = nextOrgId();
    await seedOrg(org);
    const flowA = await seedSilenceFlow(org, { thresholdMinutes: 30 });
    const flowB = await seedSilenceFlow(org, { thresholdMinutes: 30 });
    await seedPublishedAgentVersion(org, { enabled: true, pointerIds: [flowA.pointerId, flowB.pointerId] });
    const contactId = await seedContact(org);
    await seedConversation(org, contactId, 90);

    const base = silenceSweepDb();
    const tentativas: string[] = [];
    const db: SilenceSweepDb = {
      ...base,
      insertEnrollment: async (input) => {
        if (input.contact_id === contactId) tentativas.push(input.pointer_id);
        return base.insertEnrollment(input);
      },
    };
    const deps = { db, gateDb: pgGateDb(), clock: CLOCK };

    const primeira = await runSilenceSweep(deps);
    expect(primeira.enrolled).toBe(1);
    expect(tentativas).toHaveLength(1); // o 2º fluxo leu "já vivo" e pulou sem tentar
    expect(primeira.skipped_existing).toBeGreaterThanOrEqual(1);

    // os ticks seguintes (1×/min em produção) também não tentam: era aqui que
    // nasciam os ~124 mil 23505 por dia
    await runSilenceSweep(deps);
    await runSilenceSweep(deps);
    expect(tentativas).toHaveLength(1);
    expect(await countLiveForContact(org, contactId)).toBe(1);
  });

  it("contato JÁ vivo no fluxo A → sweep do fluxo B NÃO enrolla (índice do baseline)", async () => {
    const org = nextOrgId();
    await seedOrg(org);
    const flowA = await seedSilenceFlow(org, { thresholdMinutes: 30 });
    const flowB = await seedSilenceFlow(org, { thresholdMinutes: 30 });
    const agentId = await seedPublishedAgentVersion(org, {
      enabled: true,
      pointerIds: [flowA.pointerId, flowB.pointerId],
    });
    const contactId = await seedContact(org);
    await seedConversation(org, contactId, 90);

    // enrollment vivo pré-existente no fluxo A (direto, como se um sweep anterior tivesse criado)
    await pool.query(
      `insert into followup_enrollments
         (organization_id, pointer_id, version_id, contact_id, current_node_id, status, next_eval_at, agent_id)
       values ($1, $2, $3, $4, 't1', 'active', now(), $5)`,
      [org, flowA.pointerId, flowA.versionId, contactId, agentId],
    );

    const summary = await runSilenceSweep({ db: silenceSweepDb(), gateDb: pgGateDb(), clock: CLOCK });
    // fluxo B tentou, bateu no índice org-wide → skip, nada novo
    expect(await countEnrollments(flowB.pointerId, contactId)).toBe(0);
    expect(await countLiveForContact(org, contactId)).toBe(1);
    expect(summary.skipped_existing).toBeGreaterThanOrEqual(1);
  });
});

// ---- 8. Task 8.6: dedup da migration 0062 (antes do índice) -------------

describe("dedup 0062 — >1 enrollment vivo pro mesmo (org,contact) vira 1 vivo + resto cancelado", () => {
  it("mantém o started_at MAIS RECENTE; os demais → cancelled/exclusivity_backfill/next_eval_at null", async () => {
    const org = nextOrgId();
    await seedOrg(org);
    const flowA = await seedSilenceFlow(org, { thresholdMinutes: 30 });
    const flowB = await seedSilenceFlow(org, { thresholdMinutes: 30 });
    const contactId = await seedContact(org);

    try {
      // estado sujo pré-migration: 2 vivos pro mesmo (org,contact). Só é possível
      // sem o índice org-wide — reproduz um clone anterior à 0062.
      await pool.query(`drop index if exists ${ONE_LIVE}`);
      const older = await pool.query<{ id: string }>(
        `insert into followup_enrollments
           (organization_id, pointer_id, version_id, contact_id, current_node_id, status, next_eval_at, started_at)
         values ($1,$2,$3,$4,'t1','active', now(), now() - interval '2 hours') returning id`,
        [org, flowA.pointerId, flowA.versionId, contactId],
      );
      const newer = await pool.query<{ id: string }>(
        `insert into followup_enrollments
           (organization_id, pointer_id, version_id, contact_id, current_node_id, status, next_eval_at, started_at)
         values ($1,$2,$3,$4,'t1','active', now(), now() - interval '10 minutes') returning id`,
        [org, flowB.pointerId, flowB.versionId, contactId],
      );

      // A DEDUP exata da migration 0062 (window function genérica, sem ids hardcoded).
      //
      // ⚠️ ESTE PREDICADO NÃO ACOMPANHA A 0145, E É DE PROPÓSITO. Ele não
      // restaura estado: REPRODUZ o que a 0062 fazia, e a 0062 não conhecia
      // `paused_manual`. Acrescentar o status aqui mudaria o que a réplica
      // replica — o oposto de consertar. O predicado que PRECISA acompanhar as
      // migrations é o de `setOneLiveIndex`, lá em cima, porque aquele devolve o
      // banco compartilhado ao estado do baseline.
      await pool.query(
        `with ranked as (
           select id, row_number() over (
             partition by organization_id, contact_id
             order by started_at desc, id desc
           ) as rn
           from followup_enrollments
           where status in ('active','waiting_reply','paused_handoff')
         )
         update followup_enrollments e
         set status='cancelled', cancel_reason='exclusivity_backfill', next_eval_at=null, updated_at=now()
         from ranked where e.id = ranked.id and ranked.rn > 1`,
      );

      // recriar o índice único org-wide agora PASSA (dado curado antes)
      await setOneLiveIndex("organization_id, contact_id");

      expect(await countLiveForContact(org, contactId)).toBe(1);
      const kept = await pool.query<{ status: string }>(
        `select status from followup_enrollments where id = $1`,
        [newer.rows[0]!.id],
      );
      expect(kept.rows[0]!.status).toBe("active"); // o mais recente sobrevive
      const cancelled = await pool.query<{ status: string; cancel_reason: string; next_eval_at: string | null }>(
        `select status, cancel_reason, next_eval_at from followup_enrollments where id = $1`,
        [older.rows[0]!.id],
      );
      expect(cancelled.rows[0]!.status).toBe("cancelled");
      expect(cancelled.rows[0]!.cancel_reason).toBe("exclusivity_backfill");
      expect(cancelled.rows[0]!.next_eval_at).toBeNull();
    } finally {
      await pool.query(`drop index if exists ${ONE_LIVE}`);
      await pool.query(`delete from followup_enrollments where contact_id = $1`, [contactId]);
      await setOneLiveIndex("organization_id, contact_id");
    }
  });
});

// ---- pausa de reentrada e pessoa no comando (lib/followup/pausa-de-reentrada.ts) ----

/** Uma inscrição ENCERRADA deste fluxo, terminada há `agoMinutes`. */
async function seedEncerrada(
  org: string,
  pointerId: string,
  versionId: string,
  contactId: string,
  agoMinutes: number,
  status: "cancelled" | "completed" = "cancelled",
): Promise<void> {
  await pool.query(
    `insert into followup_enrollments
       (organization_id, pointer_id, version_id, contact_id, current_node_id, status, next_eval_at, completed_at, updated_at)
     values ($1, $2, $3, $4, 't1', $5, null, now() - interval '${agoMinutes} minutes', now() - interval '${agoMinutes} minutes')`,
    [org, pointerId, versionId, contactId, status],
  );
}

describe("runSilenceSweep — pausa de reentrada", () => {
  it("⭐ quem encerrou uma inscrição deste fluxo há menos que a pausa não volta a entrar", async () => {
    const org = nextOrgId();
    await seedOrg(org);
    const { pointerId, versionId } = await seedSilenceFlow(org, { thresholdMinutes: 60, reentryPauseMinutes: 48 * 60 });
    const contactId = await seedContact(org);
    await seedConversation(org, contactId, 90); // respondeu há 90 min e calou
    await seedEncerrada(org, pointerId, versionId, contactId, 90); // a resposta cancelou a inscrição

    const summary = await runSilenceSweep({ db: silenceSweepDb(), gateDb: pgGateDb(), clock: CLOCK });
    expect(summary.skipped_reentry_pause).toBe(1);
    expect(await countEnrollments(pointerId, contactId)).toBe(1); // só a encerrada
  });

  it("depois da pausa inteira sem o cliente escrever, o fluxo recomeça", async () => {
    const org = nextOrgId();
    await seedOrg(org);
    const { pointerId, versionId } = await seedSilenceFlow(org, { thresholdMinutes: 60, reentryPauseMinutes: 48 * 60 });
    const contactId = await seedContact(org);
    await seedConversation(org, contactId, 49 * 60);
    await seedEncerrada(org, pointerId, versionId, contactId, 49 * 60);

    const summary = await runSilenceSweep({ db: silenceSweepDb(), gateDb: pgGateDb(), clock: CLOCK });
    expect(summary.skipped_reentry_pause).toBe(0);
    expect(summary.enrolled).toBe(1);
    expect(await countEnrollments(pointerId, contactId)).toBe(2);
  });

  it("a mensagem mais recente do cliente recomeça a contagem, mesmo com a inscrição encerrada há mais tempo", async () => {
    const org = nextOrgId();
    await seedOrg(org);
    const { pointerId, versionId } = await seedSilenceFlow(org, { thresholdMinutes: 60, reentryPauseMinutes: 48 * 60 });
    const contactId = await seedContact(org);
    await seedEncerrada(org, pointerId, versionId, contactId, 60 * 60); // encerrada há 60 h
    await seedConversation(org, contactId, 20 * 60); // mas escreveu de novo há 20 h

    const summary = await runSilenceSweep({ db: silenceSweepDb(), gateDb: pgGateDb(), clock: CLOCK });
    expect(summary.skipped_reentry_pause).toBe(1);
    expect(await countEnrollments(pointerId, contactId)).toBe(1);
  });

  it("⭐ pausa contada do último ENVIO: encerrada há 25 h e o cliente escreveu há 20 min — entra de novo", async () => {
    // O toque curto "10 min depois de calar, no máximo 1× por dia": pela base da
    // última mensagem, a pausa de 24 h nunca se cumpriria dentro do teto de 60 min.
    const org = nextOrgId();
    await seedOrg(org);
    const { pointerId, versionId } = await seedSilenceFlow(org, {
      thresholdMinutes: 10, maxSilenceMinutes: 60, reentryPauseMinutes: 24 * 60, reentryPauseBasis: "ultimo_envio",
    });
    const contactId = await seedContact(org);
    await seedEncerrada(org, pointerId, versionId, contactId, 25 * 60);
    await seedConversation(org, contactId, 20);

    const summary = await runSilenceSweep({ db: silenceSweepDb(), gateDb: pgGateDb(), clock: CLOCK });
    expect(summary.skipped_reentry_pause).toBe(0);
    expect(summary.enrolled).toBe(1);
    expect(await countEnrollments(pointerId, contactId)).toBe(2);
  });

  it("controle da base: a mesma situação, contada da última mensagem, segue em pausa", async () => {
    const org = nextOrgId();
    await seedOrg(org);
    const { pointerId, versionId } = await seedSilenceFlow(org, {
      thresholdMinutes: 10, maxSilenceMinutes: 60, reentryPauseMinutes: 24 * 60,
    });
    const contactId = await seedContact(org);
    await seedEncerrada(org, pointerId, versionId, contactId, 25 * 60);
    await seedConversation(org, contactId, 20);

    const summary = await runSilenceSweep({ db: silenceSweepDb(), gateDb: pgGateDb(), clock: CLOCK });
    expect(summary.skipped_reentry_pause).toBe(1);
    expect(await countEnrollments(pointerId, contactId)).toBe(1);
  });

  it("pausa contada do último envio: encerrada há 2 h, ainda não entra", async () => {
    const org = nextOrgId();
    await seedOrg(org);
    const { pointerId, versionId } = await seedSilenceFlow(org, {
      thresholdMinutes: 10, maxSilenceMinutes: 60, reentryPauseMinutes: 24 * 60, reentryPauseBasis: "ultimo_envio",
    });
    const contactId = await seedContact(org);
    await seedEncerrada(org, pointerId, versionId, contactId, 2 * 60);
    await seedConversation(org, contactId, 20);

    const summary = await runSilenceSweep({ db: silenceSweepDb(), gateDb: pgGateDb(), clock: CLOCK });
    expect(summary.skipped_reentry_pause).toBe(1);
    expect(await countEnrollments(pointerId, contactId)).toBe(1);
  });

  it("controle: sem pausa configurada, o mesmo contato volta a entrar no limiar de sempre", async () => {
    const org = nextOrgId();
    await seedOrg(org);
    const { pointerId, versionId } = await seedSilenceFlow(org, { thresholdMinutes: 60 });
    const contactId = await seedContact(org);
    await seedConversation(org, contactId, 90);
    await seedEncerrada(org, pointerId, versionId, contactId, 90);

    const summary = await runSilenceSweep({ db: silenceSweepDb(), gateDb: pgGateDb(), clock: CLOCK });
    expect(summary.enrolled).toBe(1);
    expect(await countEnrollments(pointerId, contactId)).toBe(2);
  });

  it("quem nunca passou pelo fluxo entra no limiar de sempre, com pausa configurada", async () => {
    const org = nextOrgId();
    await seedOrg(org);
    const { pointerId } = await seedSilenceFlow(org, { thresholdMinutes: 60, reentryPauseMinutes: 48 * 60 });
    const contactId = await seedContact(org);
    await seedConversation(org, contactId, 90);

    const summary = await runSilenceSweep({ db: silenceSweepDb(), gateDb: pgGateDb(), clock: CLOCK });
    expect(summary.enrolled).toBe(1);
    expect(await countEnrollments(pointerId, contactId)).toBe(1);
  });
});

describe("runSilenceSweep — pessoa no comando da conversa", () => {
  it("⭐ contato em force_human não é inscrito num fluxo sem handoff_policy='allow'", async () => {
    const org = nextOrgId();
    await seedOrg(org);
    const { pointerId } = await seedSilenceFlow(org, { thresholdMinutes: 60, handoffPolicy: "cancel" });
    const contactId = await seedContact(org, { forceHuman: true });
    await seedConversation(org, contactId, 90);

    const summary = await runSilenceSweep({ db: silenceSweepDb(), gateDb: pgGateDb(), clock: CLOCK });
    expect(summary.skipped_human_owned).toBe(1);
    expect(await countEnrollments(pointerId, contactId)).toBe(0);
  });

  it("controle: com handoff_policy='allow' o fluxo foi feito para rodar com a pessoa — inscreve", async () => {
    const org = nextOrgId();
    await seedOrg(org);
    const { pointerId } = await seedSilenceFlow(org, { thresholdMinutes: 60, handoffPolicy: "allow" });
    const contactId = await seedContact(org, { forceHuman: true });
    await seedConversation(org, contactId, 90);

    const summary = await runSilenceSweep({ db: silenceSweepDb(), gateDb: pgGateDb(), clock: CLOCK });
    expect(summary.skipped_human_owned).toBe(0);
    expect(await countEnrollments(pointerId, contactId)).toBe(1);
  });
});

describe("runSilenceSweep — teto do silêncio", () => {
  it("⭐ com teto de 60 min, quem está calado há 20 min entra e quem está há 3 h fica de fora", async () => {
    const org = nextOrgId();
    await seedOrg(org);
    const { pointerId } = await seedSilenceFlow(org, { thresholdMinutes: 10, maxSilenceMinutes: 60 });
    const naFaixa = await seedContact(org);
    await seedConversation(org, naFaixa, 20);
    const antigo = await seedContact(org);
    await seedConversation(org, antigo, 180);

    await runSilenceSweep({ db: silenceSweepDb(), gateDb: pgGateDb(), clock: CLOCK });
    expect(await countEnrollments(pointerId, naFaixa)).toBe(1);
    expect(await countEnrollments(pointerId, antigo)).toBe(0);
  });
});