import { disporJobAposFalha } from "./dispor-job-apos-falha";
import { createApprovedReplyHandler } from "@/lib/agent-engine/agent/approved-reply";
import { turnKnobsFromEnv } from "@/lib/agent-engine/agent/turn-knobs";
import { createMeetDeliveryHandler } from "@/lib/agent-engine/agent/meet-delivery";
import { claimOfJob } from "@/lib/agent-engine/queue/claim";
import { AgendaDeferredError } from "@/lib/agenda/protecao-followup";
import { withServiceJob } from "@/lib/atendimento/fronteira-server";
import { StaleServiceBoundaryError } from "@/lib/atendimento/fronteira";
/**
 * Worker 24/7 do agent-engine (fusão Vendaval → DeskcommCRM) — o processo
 * long-running que o CRM não tinha: fila durável, cron/follow-up, drain do
 * event_log e os turnos do agente rico.
 *
 * Ritual de boot: env (Zod) → check do schema do harness (recusa subir sem a
 * migration 0050 aplicada — aplicar é ato de deploy) → solta órfãos → healthz →
 * loops (worker, drain, cron, holds, saúde do número).
 *
 * Graceful shutdown: SIGTERM/SIGINT → para de claimar e o prazo TOTAL começa no
 * sinal (#2823), cobrindo laços, drain e pool até SHUTDOWN_GRACE_MS: dentro do
 * prazo, fecha healthz e pool e sai 0; vencido, sai ≠ 0 sem esperar (o reaper
 * devolve o lease). Morte súbita — SIGKILL do Docker no `stop_grace_period` — é
 * o caso do reaper: lease expira e o job volta.
 *
 * Rodar: `pnpm worker` (tsx) — dev com --env-file=.env.local; container via
 * Dockerfile.worker (serviço `worker` do docker-compose).
 */

// Sentry precisa iniciar antes de qualquer import de lógica de negócio: os
// imports abaixo são avaliados em ordem textual, e cada `import` só avança
// para o próximo depois de o módulo importado terminar de rodar — então
// colocar o `Sentry.init` aqui (antes dos imports do agent-engine) garante
// que ele está de pé antes de qualquer código do worker executar.
//
// Mesma lógica de `sentry.server.config.ts`/`sentry.edge.config.ts`
// (reaproveitada, não duplicada): DSN resolvido por `resolveSentryDsn`,
// amostragem de trace condicionada ao Sentry da comunidade via
// `isCommunityDsn` (issue #100), e a coleta restrita + scrub de
// `lib/sentry/privacidade.ts`.
// O `@sentry/nextjs` funciona fora do Next — aqui é só `Sentry.init` puro,
// sem `instrumentation.ts` porque o worker não é um processo Next.
import * as Sentry from "@sentry/nextjs";
import { resolveSentryDsn, isCommunityDsn, DEFAULT_SENTRY_DSN } from "@/lib/sentry/dsn";
import { opcoesDePrivacidade } from "@/lib/sentry/privacidade";

const sentryDsn = resolveSentryDsn(process.env.SENTRY_DSN);
const sentryCommunity = isCommunityDsn(sentryDsn);

Sentry.init({
  dsn: sentryDsn,

  // No Sentry da comunidade, só erro (issue #100). Ver isCommunityDsn().
  tracesSampleRate: sentryCommunity ? 0 : 1,

  // Coleta restrita + scrub, num ponto só (Sentry 11 coleta amplo por default).
  ...opcoesDePrivacidade,
});

// Transparência de telemetria (mesma mensagem de sentry.server.config.ts,
// adaptada para o processo worker): uma linha no boot dizendo o que está
// ativo e como desligar.
if (!sentryDsn) {
  console.info("[telemetria] worker: Desligada (SENTRY_DSN=off) — nenhum erro é enviado.");
} else if (sentryDsn === DEFAULT_SENTRY_DSN) {
  console.info(
    "[telemetria] worker: Relatórios de erro anonimizados ATIVOS (Sentry da comunidade). " +
      "Sem rastreamento de performance nem replay de sessão. " +
      "Desligue com SENTRY_DSN=off, ou envie pro seu com SENTRY_DSN=<seu-dsn>.",
  );
} else {
  console.info("[telemetria] worker: Erros sendo enviados ao Sentry configurado em SENTRY_DSN.");
}

import { hostname } from "node:os";

import type pg from "pg";

import { createInboundTurnHandler } from "@/lib/agent-engine/agent/inbound-turn";
import {
  createFollowupTurnHandler,
  type FollowupTurnDeps,
} from "@/lib/agent-engine/agent/followup-turn";
import { createCaseReplyTurnHandler } from "@/lib/agent-engine/agent/case-reply-turn";
import { createOperatorTurnHandler } from "@/lib/agent-engine/agent/operator-turn";
import { completeTurnForEnrollment, createPgAdminClient } from "@/lib/followup/turn-bridge";
import { seedPlatformPlaybook } from "@/lib/agent-engine/agent/playbook-seed";
import { runCronLoop } from "@/lib/agent-engine/cron/scheduler";
import { createPool } from "@/lib/agent-engine/db/pool";
import {
  carregarComportamentoPorPool,
  pisoDoComportamentoDoMotor,
} from "@/lib/instalacao/comportamento-sql";
import { prontidaoDoDrainDaIa, runDrainLoop } from "@/lib/agent-engine/edge/crm/drain";
import { sincronizarAvisoDoDrain } from "@/lib/agent-engine/edge/crm/vigilancia-do-drain";
import { runEventLogDrainLoop } from "@/lib/event-log/drain-loop";
import { crmEdgeConfigFromEnv } from "@/lib/agent-engine/edge/crm/mcp-client";
import { enforceHolds } from "@/lib/agent-engine/edge/crm/session-watchdog";
import { runVoiceCallsBridgeLoop } from "@/lib/wacalls/events-bridge";
import { runSessionWatchdogLoop } from "@/lib/agent-engine/edge/crm/session-reconciler";
import { runHealthLoop } from "@/lib/agent-engine/health/circuit";
import { runFlywheelLoop } from "@/lib/agent-engine/flywheel/live";
import { llmEdgeConfigFromEnv } from "@/lib/agent-engine/edge/llm/run-model-call";
import { loadEnv, type Env } from "@/lib/agent-engine/env";
import { urlDoSupabaseNoServidor } from "@/lib/supabase/url-do-servidor";
import { createLogger, type Logger } from "@/lib/agent-engine/obs/logger";
import {
  evaluateCacheHitAlert,
  recordRunMetrics,
  type CacheAlertKnobs,
} from "@/lib/agent-engine/obs/metrics";
import { rodarLoopDaFila } from "@/lib/agent-engine/queue/loop";
import { criarSonecaDaFila } from "@/lib/agent-engine/queue/soneca";
import { createHealthzServer } from "./healthz";
import { desligarNoPrazo } from "./desligamento-no-prazo";
import {
  adiarAteORecarregar,
  avisarFaltaDeSaldo,
  deveEsperarSaldo,
  encerrarAvisoDeFaltaDeSaldo,
  esperouPorSaldo,
  jaNaoHaOQueResponder,
} from "@/lib/agent-engine/queue/espera-de-saldo";
import {
  claimJobs,
  completeJob,
  faltaParaOProximoJob,
  reapExpiredJobs,
  type JobKind,
  type JobRow,
} from "@/lib/agent-engine/queue/queue";

export interface JobHandlerContext {
  workerId: string;
}
export type JobHandler = (job: JobRow, pool: pg.Pool, ctx: JobHandlerContext) => Promise<void>;

/** 1ª linha, truncada — PII fora de log. */
function errMsg(err: unknown): string {
  const message = err instanceof Error ? err.message : String(err);
  return (message.split("\n", 1)[0] ?? "").slice(0, 300);
}

/**
 * VETO PERMANENTE DE NEGÓCIO × INCIDENTE DE SISTEMA — a fila precisa distinguir.
 *
 * O erro é lido pela PROPRIEDADE `terminal`, e não pela classe, porque o
 * contrato é da fila e não do seam que o produziu: quem sabe que "tentar de novo
 * daqui a um minuto dá o mesmo resultado" é quem lança. Hoje o único produtor é
 * `LlmBudgetExceededError` (`lib/agent-engine/edge/llm/run-model-call.ts`), que
 * declara a propriedade com a razão escrita ao lado.
 *
 * Sem esta distinção, um bloqueio por orçamento ia para `failJob`, que reagenda
 * até `max_attempts` (5) e então insere um `job_dead` **crítico por job, sem
 * dedup**: N conversas × 5 tentativas viravam N alertas críticos rotulados "Uma
 * tarefa do assistente falhou", afogando o único `budget_exceeded` — que é o
 * alerta que explica. O precedente está escrito no próprio `queue.ts`: "Caso
 * real desta VPS: 16 alertas críticos idênticos".
 *
 * ⚠️ ESTA DECISÃO SÓ É SEGURA PORQUE O TRABALHO NÃO FICA ÓRFÃO — e essa parte
 * mora em OUTRO arquivo, então ela é dita aqui com o escopo exato:
 *
 *   * `inbound_turn`, `followup_turn` e `case_reply_turn` passam por
 *     `runAgentTurn` (`lib/agent-engine/agent/inbound-turn.ts`), que envolve o
 *     turno INTEIRO em `comHandoffSeOrcamentoAcabar`. Qualquer chamada de modelo
 *     do turno — inclusive as indiretas (`classifyStage`, `maybeCompact`), que
 *     rodam ANTES da principal — cai na escolta, e a conversa é devolvida à fila
 *     humana antes do relance. Nesses três, o job termina porque OUTRA PESSOA
 *     assumiu, não porque foi descartado. (A versão anterior desta frase era
 *     falsa: a escolta cobria só as duas chamadas diretas, o classificador
 *     estourava primeiro, e este `cancelJob` descartava o job com o lead no
 *     vácuo — a frase falsa era a justificativa da decisão irreversível.)
 *   * `operator_turn` NÃO tem handoff, e não deve ter: o lead já foi respondido
 *     pelo Conversador, e o Operador é retaguarda. O que se perde ali é o
 *     `registrarDesfecho` daquele turno — perda real, limitada e sem conserto
 *     por retry (o teto continua estourado no minuto seguinte). O sinal que
 *     sobra é o `budget_exceeded` na Central, que é o alerta que explica.
 */
function ehVetoPermanenteDeNegocio(err: unknown): boolean {
  return (
    typeof err === "object" && err !== null && (err as { terminal?: unknown }).terminal === true
  );
}

/**
 * O boot NÃO aplica migrations (ato de deploy, via supabase/migrations + kit) —
 * só confere que o schema do harness existe e recusa subir sem ele.
 */
async function assertHarnessSchema(pool: pg.Pool): Promise<void> {
  const sentinels = ["job_queue", "lead_checkpoints", "agent_inbox_items", "send_ledger"];
  const { rows } = await pool.query<{ missing: string }>(
    `select t.name as missing
     from unnest($1::text[]) as t(name)
     where to_regclass('public.' || t.name) is null`,
    [sentinels],
  );
  if (rows.length > 0) {
    throw new Error(
      `schema do harness ausente no banco (tabelas: ${rows.map((r) => r.missing).join(", ")}) — aplique a migration 0050_agent_harness antes de subir o worker`,
    );
  }
}

/**
 * O `/healthz` + `/metrics` do worker moram em `./healthz` desde a #2505: o
 * `main()` no topo deste módulo impede importá-lo num teste, e o caso de aceite
 * da issue precisa subir o servidor de verdade para provar o 503 com o laço
 * travado.
 */

export async function startWorker(
  env: Env,
  handlers: Map<JobKind, JobHandler>,
  log: Logger = createLogger(),
): Promise<void> {
  const pool = createPool(env.SUPABASE_DB_URL, (err) =>
    log.error("pool: conexão caiu — recria no próximo uso", { error: errMsg(err) }),
  );
  const workerId = `agent-engine-${hostname()}-${process.pid}`;

  await assertHarnessSchema(pool);

  // Self-host limpo: sem ponteiro platform, TODO inbound_turn morre. O seed só
  // age quando não existe ponteiro nenhum (regra dura nº 10 — nunca move
  // ponteiro existente) e é concorrência-safe (advisory lock).
  const playbookSeed = await seedPlatformPlaybook(pool);
  if (playbookSeed === "seeded") {
    log.info("playbook platform seedado no boot (primeiro boot do self-host)");
  }

  const bootReap = await reapExpiredJobs(pool, {
    visibilityTimeoutMs: env.QUEUE_VISIBILITY_TIMEOUT_MS,
  });
  if (bootReap.revived + bootReap.dead > 0) {
    log.warn("órfãos soltos no boot", bootReap);
  }

  // O comportamento da INSTALAÇÃO entra no processo ANTES dos laços que
  // consomem turnos: a releitura abaixo só acontece no primeiro tique do reaper
  // (QUEUE_REAPER_INTERVAL_MS, 60 s por padrão), e até lá o orçamento e os knobs
  // do turno responderiam com o piso do `.env`, não com a escolha da tela.
  // Nunca lança: sem leitura boa, vale o piso — o comportamento de antes.
  await carregarComportamentoPorPool(pool, pisoDoComportamentoDoMotor(env));

  const server = createHealthzServer(pool, log, env.METRICS_WINDOW_MS);
  await new Promise<void>((resolve, reject) => {
    server.once("error", reject);
    server.listen(env.HEALTH_PORT, "0.0.0.0", resolve);
  });
  const address = server.address();
  const port = typeof address === "object" && address !== null ? address.port : env.HEALTH_PORT;

  let shuttingDown = false;
  const inFlight = new Set<Promise<void>>();

  const reaperTimer = setInterval(() => {
    // Recarrega o comportamento da INSTALAÇÃO no ritmo do reaper: é o que faz
    // uma escolha feita na tela de admin alcançar ESTE processo sem restart
    // (issue #1034). O memo de 30 s evita ir ao banco a cada tique, e o
    // carregador nunca lança — o `.catch` cobre o caso impossível sem deixar
    // rejeição solta (o worker morre com promise rejeitada não tratada).
    carregarComportamentoPorPool(pool, pisoDoComportamentoDoMotor(env)).catch((err: unknown) =>
      log.error("comportamento da instalação: releitura falhou", { error: errMsg(err) }),
    );
    reapExpiredJobs(pool, { visibilityTimeoutMs: env.QUEUE_VISIBILITY_TIMEOUT_MS })
      .then((reaped) => {
        if (reaped.revived + reaped.dead > 0) log.warn("reaper devolveu jobs órfãos", reaped);
      })
      .catch((err: unknown) => log.error("reaper falhou", { error: errMsg(err) }));
    // O AVISO DE LAÇO PARADO (#2505) sai do relógio do reaper, e não do laço:
    // um laço travado não pode ser quem avisa de si mesmo. No caso saudável o
    // custo é um UPDATE que casa zero linhas, e `sincronizarAvisoDoDrain` nunca
    // lança — falha da Central vira `warn`.
    void sincronizarAvisoDoDrain(
      pool,
      prontidaoDoDrainDaIa().parado ? "parado" : "saudavel",
      log,
    );
  }, env.QUEUE_REAPER_INTERVAL_MS);

  // Holds de sessão/saúde: retém jobs de envio de número fora do ar (WORKING é a
  // fonte channel_sessions, mantida pelo webhook do WAHA) — ritmo do reaper serve.
  const holdsTimer = setInterval(() => {
    enforceHolds(pool)
      .then(({ held, released }) => {
        if (held + released > 0) log.info("holds de sessão aplicados", { held, released });
      })
      .catch((err: unknown) => log.error("enforceHolds falhou", { error: errMsg(err) }));
  }, env.QUEUE_REAPER_INTERVAL_MS);

  const loopsAbort = new AbortController();

  // A soneca do laço da fila (#3688). Nasce aqui porque DOIS laços a usam: o da
  // fila dorme nela, e o drain a acorda ao enfileirar um turno — mesmo worker,
  // promessa local, sem polling novo. O `signal` é o que faz o `docker stop`
  // cortar a soneca em curso na hora.
  const sonecaDaFila = criarSonecaDaFila({ signal: loopsAbort.signal });

  // Drain do event_log (mesmo banco pós-fusão) — transforma dispatch_requested em
  // jobs. Fase 0 (convergência, spec 2026-07-23): o drain liga SEMPRE — o
  // dispatcher nativo EPIC-13 foi aposentado, o engine é o único consumidor.
  if (env.AGENT_DISPATCH_CONSUMER === "native") {
    log.warn(
      "AGENT_DISPATCH_CONSUMER=native é OBSOLETO (Fase 0) — o drain do engine é o único consumidor; valor ignorado",
      {},
    );
  }
  const drainLoop = runDrainLoop(
    pool,
    {
      batchSize: env.CRM_DRAIN_BATCH_SIZE,
      intervalMs: env.CRM_DRAIN_INTERVAL_MS,
      idleIntervalMs: env.CRM_DRAIN_IDLE_INTERVAL_MS,
      debounceMs: env.INBOUND_DEBOUNCE_MS,
      reapTimeoutMs: env.CRM_EVENT_REAP_TIMEOUT_MS,
      allowlistTtlMs: env.AI_ALLOWLIST_TTL_DAYS * 24 * 60 * 60 * 1000,
      // Quem corta a soneca do laço da fila quando um turno nasce (#3688).
      acordarFila: () => sonecaDaFila.acordar(),
    },
    log,
    loopsAbort.signal,
  );

  // Handlers do event_log (mídia, branding, follow-up…) no ritmo do worker.
  // Antes disto eles só rodavam pelo cron `event-log-drain`, 1×/min: a cadeia
  // persist→derive de um áudio levava 103-188s contra os 120s que o drain do
  // turno espera, e o agente respondia "não consigo ouvir" com a transcrição
  // pronta segundos depois. O cron continua como rede de segurança — o claim
  // otimista do `drainEventLog` torna os dois seguros em paralelo.
  const eventLogLoop = runEventLogDrainLoop(
    {
      intervalMs: env.EVENT_LOG_DRAIN_INTERVAL_MS,
      idleIntervalMs: env.EVENT_LOG_DRAIN_IDLE_INTERVAL_MS,
      idleMaxMs: env.EVENT_LOG_DRAIN_IDLE_MAX_INTERVAL_MS,
      staleSweepMs: env.EVENT_LOG_DRAIN_STALE_SWEEP_MS,
      batchSize: env.EVENT_LOG_DRAIN_BATCH_SIZE,
    },
    log,
    loopsAbort.signal,
  );

  // Watchdog de sessão (4A-2): reconcilia channel_sessions×WAHA + redrive de
  // queued. Liga só com as credenciais do WAHA no env (sem elas: warn + off).
  const sessionWatchdogLoop =
    env.WAHA_API_BASE_URL !== undefined && env.WAHA_API_KEY !== undefined
      ? runSessionWatchdogLoop(
          pool,
          {
            wahaBaseUrl: env.WAHA_API_BASE_URL,
            wahaApiKey: env.WAHA_API_KEY,
            intervalMs: env.WATCHDOG_INTERVAL_MS,
            redriveMinAgeMs: env.WATCHDOG_REDRIVE_MIN_AGE_MS,
            redriveBatchSize: env.WATCHDOG_REDRIVE_BATCH_SIZE,
            redriveSpacingMs: env.WATCHDOG_REDRIVE_SPACING_MS,
          },
          log,
          loopsAbort.signal,
        )
      : (log.warn("watchdog de sessão OFF — WAHA_API_BASE_URL/WAHA_API_KEY ausentes no env", {}),
        Promise.resolve());

  // Ponte de eventos WaCalls (spec 18, §4.2) — chamada de voz, opt-in por
  // org. Sem a env, fica OFF: instalação que não usa a feature não paga o
  // custo de uma conexão SSE tentando alcançar um serviço que não existe.
  const voiceCallsBridgeLoop =
    env.WACALLS_API_BASE_URL !== undefined && env.WACALLS_API_TOKEN !== undefined
      ? runVoiceCallsBridgeLoop(
          pool,
          {
            baseUrl: env.WACALLS_API_BASE_URL,
            apiToken: env.WACALLS_API_TOKEN,
            maxBackoffMs: env.WACALLS_BRIDGE_MAX_BACKOFF_MS,
          },
          log,
          loopsAbort.signal,
        )
      : (log.info('ponte WaCalls OFF — endereço ou credencial ausente no env', {}), Promise.resolve());

  // Circuito de saúde do número (block/response rate → hold).
  const healthLoop = runHealthLoop(
    pool,
    { intervalMs: env.NUMBER_HEALTH_INTERVAL_MS },
    log,
    loopsAbort.signal,
  );

  // Flywheel agendado (4B): judge→distiller periódico sobre turnos reais.
  // Precisa da camada LLM; sem intervalo (0) fica OFF.
  const flywheelLoop =
    env.FLYWHEEL_INTERVAL_MS > 0
      ? runFlywheelLoop(
          pool,
          llmEdgeConfigFromEnv(env),
          { intervalMs: env.FLYWHEEL_INTERVAL_MS, limit: env.FLYWHEEL_BATCH_LIMIT, log },
          loopsAbort.signal,
        )
      : Promise.resolve();

  // Cron persistente por contato (follow-up) — só enfileira em job_queue.
  const cronLoop = runCronLoop(
    pool,
    {
      intervalMs: env.CRON_TICK_INTERVAL_MS,
      batchSize: env.CRON_BATCH_SIZE,
      staggerWindowMs: env.CRON_STAGGER_WINDOW_MS,
      retryBaseMs: env.CRON_RETRY_BASE_MS,
    },
    log,
    loopsAbort.signal,
  );

  const cacheAlertKnobs: CacheAlertKnobs = {
    windowMs: env.METRICS_WINDOW_MS,
    cacheHitAlertThreshold: env.CACHE_HIT_ALERT_THRESHOLD,
    cacheHitAlertMinRuns: env.CACHE_HIT_ALERT_MIN_RUNS,
  };

  const runJob = async (job: JobRow): Promise<void> => {
    try {
      const handler = handlers.get(job.kind);
      if (!handler) {
        throw new Error(`nenhum handler registrado para kind=${job.kind}`);
      }
      // Entrega transacional liquida estado do compromisso + job na mesma RPC cercada.
      if (job.kind === "transactional_delivery" || job.kind === "approved_reply") {
        await handler(job, pool, { workerId });
        return;
      }
      // Voltou de uma espera de saldo (`espera-de-saldo.ts`): se, enquanto a IA
      // esperava, alguém do nosso lado já respondeu o cliente, não há o que
      // responder — sair agora seria repetir quem já atendeu.
      if (esperouPorSaldo(job) && (await jaNaoHaOQueResponder(pool, job))) {
        await completeJob(pool, job.id, workerId, undefined, claimOfJob(job)?.acquired_at);
        log.info("job encerrado: a conversa foi respondida enquanto a IA esperava saldo", {
          job_id: job.id,
          kind: job.kind,
        });
        return;
      }
      await withServiceJob(pool, job, () => handler(job, pool, { workerId }));
      await completeJob(pool, job.id, workerId, undefined, claimOfJob(job)?.acquired_at);
      log.info("job concluído", { job_id: job.id, kind: job.kind });
      if (esperouPorSaldo(job)) {
        try {
          await encerrarAvisoDeFaltaDeSaldo(pool, job.organization_id);
        } catch (avisoErr) {
          log.error("aviso de falta de saldo não foi encerrado", { job_id: job.id, error: errMsg(avisoErr) });
        }
      }
      try {
        const llmCalls = await recordRunMetrics(pool, job);
        if (llmCalls > 0) {
          await evaluateCacheHitAlert(pool, job.organization_id, cacheAlertKnobs);
        }
      } catch (metricsErr) {
        log.error("métricas do run não registradas", { job_id: job.id, error: errMsg(metricsErr) });
      }
    } catch (err) {
      if (job.kind === "transactional_delivery" || job.kind === "approved_reply") {
        // O consumer é dono do settle atômico; falha de armazenamento ou claim
        // ausente deixa a aquisição ao reaper, sem fallback por locked_by.
        log.error("entrega transacional aguarda recuperação da fila", {
          job_id: job.id,
          code:
            err instanceof StaleServiceBoundaryError
              ? "claim_missing_or_stale"
              : "settle_unavailable",
        });
        return;
      }
      if (err instanceof AgendaDeferredError && err.protection.motivo !== "leitura_indisponivel") {
        await pool.query(
          "update job_queue set status='pending',run_after=$4,locked_by=null,locked_at=null,attempts=greatest(attempts-1,0),last_error=$5 where id=$1 and organization_id=$2 and locked_by=$3 and status='running' and locked_at=$6",
          [
            job.id,
            job.organization_id,
            workerId,
            err.protection.reavaliar_em,
            err.message,
            claimOfJob(job)?.acquired_at ?? null,
          ],
        );
        return;
      }
      // Conta do provedor sem crédito: a resposta espera a recarga, sem gastar
      // tentativa, até o teto de `espera-de-saldo.ts` — e a Central diz a causa.
      // Passado o teto, segue o caminho comum abaixo (`failJob` → `job_dead`).
      if (deveEsperarSaldo(job, err)) {
        log.warn("provedor de IA sem saldo — resposta esperando a recarga", {
          job_id: job.id,
          kind: job.kind,
          error: errMsg(err),
        });
        try {
          await avisarFaltaDeSaldo(pool, job.organization_id, err);
        } catch (avisoErr) {
          log.error("aviso de falta de saldo não entrou na Central", { job_id: job.id, error: errMsg(avisoErr) });
        }
        try {
          await adiarAteORecarregar(pool, job, workerId, err, claimOfJob(job)?.acquired_at);
        } catch (adiarErr) {
          log.error("espera de saldo indisponível — lease expira via reaper", {
            job_id: job.id,
            error: errMsg(adiarErr),
          });
          Sentry.captureException(adiarErr);
        }
        return;
      }
      const terminal = err instanceof StaleServiceBoundaryError || ehVetoPermanenteDeNegocio(err);
      // `warn`, não `error`: veto de negócio é o sistema funcionando como
      // configurado. Rotulá-lo de erro treina quem opera a ignorar o painel.
      if (terminal) {
        log.warn("job cancelado por veto permanente de negócio", {
          job_id: job.id,
          kind: job.kind,
          error: errMsg(err),
        });
      } else {
        log.error("job falhou", { job_id: job.id, kind: job.kind, error: errMsg(err) });
        // Só o caminho não-terminal: veto de negócio (acima) é o sistema
        // funcionando como configurado, não um incidente — mandar isso pro
        // Sentry afogaria o painel com o mesmo ruído que `ehVetoPermanenteDeNegocio`
        // existe para evitar (ver comentário da função, "16 alertas críticos
        // idênticos"). Isto aqui é falha real e não tratada do turno/job.
        Sentry.captureException(err);
      }
      try {
        await disporJobAposFalha(pool, job, workerId, err, terminal, log);
      } catch (failErr) {
        log.error("disposição do job indisponível — lease expira via reaper", {
          job_id: job.id,
          error: errMsg(failErr),
        });
        // Falha dupla: o job falhou E a fila não conseguiu registrar a falha
        // (banco fora do ar, tipicamente). O job só se recupera pelo reaper —
        // vale saber que isto aconteceu.
        Sentry.captureException(failErr);
      }
    }
  };

  const workerLoop = rodarLoopDaFila<JobRow>({
    relogio: () => faltaParaOProximoJob(pool),
    claimar: () => claimJobs(pool, { workerId, maxConcurrency: env.QUEUE_MAX_CONCURRENCY }),
    aoClaimar: (jobs) => {
      for (const job of jobs) {
        const running = runJob(job);
        inFlight.add(running);
        void running.finally(() => inFlight.delete(running));
      }
    },
    // O `{ signal }` (dentro da soneca) é o que faz o `docker stop` não esperar a
    // espera inteira — e é por isso que `rodarLoopDaFila` trata a rejeição em vez
    // de propagá-la: `sleep` abortado REJEITA, e essa rejeição chegaria ao `await
    // workerLoop` logo abaixo, matando o processo antes do drain dos jobs em voo.
    soneca: sonecaDaFila,
    deveParar: () => shuttingDown,
    intervalos: {
      ociosoMs: env.QUEUE_POLL_INTERVAL_MS,
      retryMs: env.QUEUE_CLAIM_RETRY_INTERVAL_MS,
      ociosoMaxMs: env.QUEUE_IDLE_MAX_INTERVAL_MS,
    },
    log,
  });

  let resolveStopped!: () => void;
  const stopped = new Promise<void>((resolve) => {
    resolveStopped = resolve;
  });

  const shutdown = async (signal: NodeJS.Signals): Promise<void> => {
    if (shuttingDown) return;
    shuttingDown = true;
    // (#2823) O prazo TOTAL arma no SINAL e cobre laços, drain e pool. Antes, a
    // contagem começava DEPOIS de `await Promise.all(lacos)` e de
    // `await workerLoop`, e `pool.end()` ficava fora dela: um laço preso (o do
    // `event_log` numa chamada de rede, sem PostgREST) levou a parada ociosa a
    // 19,6 s MEDIDOS por fora de qualquer prazo — e o que sobrava ainda competia
    // com o SIGKILL do Docker, que chega no `stop_grace_period`. O porquê de o
    // relógio ter de armar aqui está em `./desligamento-no-prazo.ts`.
    const fim = await desligarNoPrazo({
      prazoMs: env.SHUTDOWN_GRACE_MS,
      sinal: signal,
      emVoo: () => inFlight.size,
      aoSinal: () => {
        clearInterval(reaperTimer);
        clearInterval(holdsTimer);
        server.close();
        server.closeIdleConnections();
        loopsAbort.abort();
      },
      // A ORDEM importa: os laços param de claimar antes de o drain ser
      // aguardado, e um claim já iniciado termina dentro de `workerLoop` (não
      // larga lease no meio). Tudo isto é o que o prazo tem de cobrir.
      encerrar: async () => {
        await Promise.all([
          drainLoop,
          eventLogLoop,
          healthLoop,
          cronLoop,
          sessionWatchdogLoop,
          flywheelLoop,
          voiceCallsBridgeLoop,
        ]);
        await workerLoop;
        await Promise.all([...inFlight]);
        await pool.end();
      },
      log,
    });
    if (fim === "prazo-estourado") {
      process.exit(1);
    }
    resolveStopped();
  };
  process.once("SIGTERM", (signal) => void shutdown(signal));
  process.once("SIGINT", (signal) => void shutdown(signal));

  // A linha que a Fase 0 exige ver: worker conectado ao Supabase, schema ok, pronto.
  // Os três intervalos vão junto porque governam a conta de egress do banco (issues
  // #258 e #3688) e quem os ajusta precisa conseguir CONFERIR o que está em vigor sem
  // ler o código-fonte — foi o que os autores das issues tiveram de fazer para achar
  // o knob.
  log.info("agent-engine pronto", {
    worker_id: workerId,
    healthz_port: port,
    max_concurrency: env.QUEUE_MAX_CONCURRENCY,
    poll_ocioso_ms: env.QUEUE_POLL_INTERVAL_MS,
    // Igual ao piso = recuo desligado (o default, doc 242).
    poll_ocioso_teto_ms: env.QUEUE_IDLE_MAX_INTERVAL_MS ?? env.QUEUE_POLL_INTERVAL_MS,
    claim_retry_ms: env.QUEUE_CLAIM_RETRY_INTERVAL_MS,
  });
  // Acima de 10 s a espera ociosa passa do idleTimeoutMillis do pool (10 s, o
  // default do pg): a conexão morre entre uma rodada e outra, e cada consulta ao
  // relógio volta a pagar TCP+TLS+startup. Medido: 499 B por rodada depois de 12 s
  // de sono contra 66 B depois de 2 s — subir o intervalo além daí gasta mais do
  // que economiza. Aviso em vez de teto no schema: um `.max()` faria o worker
  // RECUSAR subir numa máquina cujo .env já tem um valor maior, e a doutrina de
  // packaging proíbe atualização que exija edição manual de arquivo.
  if (env.QUEUE_POLL_INTERVAL_MS >= 10_000) {
    log.warn(
      "QUEUE_POLL_INTERVAL_MS ≥ 10s — cada rodada ociosa reconecta ao banco e gasta MAIS que um valor menor",
      {
        poll_ocioso_ms: env.QUEUE_POLL_INTERVAL_MS,
      },
    );
  }
  // Teto abaixo do piso (#3878): o laço grampeia no piso — recuo nunca acelera —
  // mas o operador precisa saber que o valor que ele escreveu não está em vigor.
  // Sem esta linha, `QUEUE_IDLE_MAX_INTERVAL_MS=500` com piso 2000 pareceria
  // "recuo ligado" e não faria nada.
  if (
    env.QUEUE_IDLE_MAX_INTERVAL_MS !== undefined &&
    env.QUEUE_IDLE_MAX_INTERVAL_MS < env.QUEUE_POLL_INTERVAL_MS
  ) {
    log.warn(
      "QUEUE_IDLE_MAX_INTERVAL_MS abaixo de QUEUE_POLL_INTERVAL_MS — o piso manda: o recuo não acelera a fila",
      {
        poll_ocioso_ms: env.QUEUE_POLL_INTERVAL_MS,
        poll_ocioso_teto_ms: env.QUEUE_IDLE_MAX_INTERVAL_MS,
      },
    );
  }
  await stopped;
}

export async function main(): Promise<void> {
  const env = loadEnv();
  const log = createLogger();
  const handlers = new Map<JobKind, JobHandler>();
  const turnDeps: FollowupTurnDeps = {
    crmCfg: crmEdgeConfigFromEnv({
      SUPABASE_URL: urlDoSupabaseNoServidor(env.SUPABASE_SERVER_URL, env.NEXT_PUBLIC_SUPABASE_URL),
      SUPABASE_SERVICE_ROLE_KEY: env.SUPABASE_SERVICE_ROLE_KEY,
    }),
    llmCfg: llmEdgeConfigFromEnv(env),
    knobs: turnKnobsFromEnv(env),
    log,
    // Onda 5 (Task 5.1): fecha o turno dirigido por fluxo de volta no enrollment —
    // o worker fala pg puro (nunca Supabase client), então usa o adapter pg de
    // lib/followup/turn-bridge.ts (equivalente ao createSupabaseAdminClient das
    // rotas Next.js, mas pra este processo).
    completeFollowupTurn: (
      pool,
      { organizationId, enrollmentId, nodeId, jobId, jobClaim, result },
    ) =>
      completeTurnForEnrollment(
        createPgAdminClient(pool),
        organizationId,
        enrollmentId,
        nodeId,
        result,
        undefined,
        jobId,
        jobClaim,
      ),
  };
  handlers.set("approved_reply", createApprovedReplyHandler(turnDeps));
  handlers.set("transactional_delivery", createMeetDeliveryHandler(turnDeps));
  handlers.set("inbound_turn", createInboundTurnHandler(turnDeps));
  handlers.set("followup_turn", createFollowupTurnHandler(turnDeps));
  handlers.set("case_reply_turn", createCaseReplyTurnHandler(turnDeps));
  // Spec 16 §3.2 — o papel OPERADOR. Registrado sempre; quem decide se ele age é
  // `operator_enabled` na versão publicada (default false), lido a cada job. Um
  // worker que não conhecesse o kind faria os jobs morrerem em 'dead' sem que
  // ninguém entendesse por quê.
  handlers.set("operator_turn", createOperatorTurnHandler(turnDeps));
  await startWorker(env, handlers, log);
}

// tsx roda este arquivo como entrypoint direto.
main().catch((err: unknown) => {
  process.stderr.write(`boot falhou: ${errMsg(err)}\n`);
  Sentry.captureException(err);
  // `captureException` só enfileira — sem o flush, o `process.exit` teria boa
  // chance de matar o processo antes do envio sair (é exatamente o cenário:
  // boot falhou, o processo já está de saída). 2s é o mesmo teto usado pelo
  // `SHUTDOWN_GRACE_MS` default deste worker para não pendurar o exit.
  void Sentry.flush(2000).finally(() => process.exit(1));
});