diff --git a/TODO.md b/TODO.md index 398e866..d9a59de 100644 --- a/TODO.md +++ b/TODO.md @@ -497,11 +497,60 @@ platform admin cria/apaga, BYOK isolado por RLS (tenant B nunca vê BYOK do tenant A, 404 em id direto), modelo de provider GLOBAL visível dos dois tenants, redactor com 5 tipos de dado sensível -- [ ] Pipeline assíncrono pós-CALL_ENDED, transcrição/análise/prompts/ - scorecards/usage metering reais — PHASE 20 +- [x] Pipeline assíncrono pós-CALL_ENDED, transcrição/análise/prompts — + PHASE 20 (scorecards/usage metering completo ficam pra PHASE 21) -## PHASE 20+ — ver `agente.md` seções 104 em diante (Transcrição, Análise, -Scorecards, Usage Metering, Billing, Frontend, Security, Tests) +## PHASE 20 — IA: Pipeline (agente.md secao 104-116) — ver docs/AI_PIPELINE.md +- [x] `packages/ai`: `privacy.ts` (cascata Tenant>Queue>Campaign, secao + 122), `retry.ts` (backoff exponencial + dead-letter, secao 108), + `call-analysis-schema.ts` (schema JSON + validação manual do + resultado, secao 112-113), `wav-stereo-split.ts` (parser WAV RIFF + próprio, separa estéreo em 2 mono pra transcrever cada canal + independente, sem depender de diarização probabilística nem ffmpeg) +- [x] `apps/freeswitch-events/src/ai-trigger.ts`: decide (entitlement do + Plan + cascata de privacidade, as duas precisam passar) se uma + gravação recém-`AVAILABLE` vira um `AIJob(TRANSCRIPTION)` — + encadeado em `recording.ts` logo após `Recording.create` +- [x] `packages/entitlements`: `isFeatureEnabled` (versão não-lançante de + `assertFeatureEnabled`, pra decisões em background) +- [x] `apps/ai-worker` (serviço novo, Docker): tick poll com `FOR UPDATE + SKIP LOCKED` (mesmo padrão de `lead-reservation.ts`), resolve + provider/modelo (BYOK do tenant > GLOBAL da plataforma, primeira + capability compatível), processa TRANSCRIPTION (baixa do object + storage, separa estéreo, transcreve os 2 canais, persiste + `CallTranscription`+`CallTranscriptSegment`+`AIUsageRecord`, + encadeia ANALYSIS se a privacidade permitir) e ANALYSIS (resolve + `AIPromptTemplate` com override de Campaign, redige dados sensíveis + SEMPRE antes de mandar pro provider — independente do nível de + privacidade, que só controla SE roda —, valida o resultado, persiste + `CallAIAnalysis`+3 `AIUsageRecord`) +- [x] `apps/api/src/ai/ai-prompts.controller.ts`: CRUD de + `AIPromptTemplate`/`AIPromptVersion` (secao 114-116), mesma RLS + híbrida GLOBAL/tenant dos providers/models; versões nunca editadas + in-place, `activeVersionId` aponta pra qual está em uso +- [x] Testado ponta a ponta contra o `b2bcall-ai-worker` real (não mocks) + e Postgres real com RLS: cascata de privacidade+entitlement em 3 + cenários reais (permissivo→job criado, tenant AI_OFF→nada, + privacidade OK mas Plan sem IA→nada), WAV sintético real gravado no + object storage + `Recording` real + `AIJob` real reservado via SKIP + LOCKED pelo worker, download+split+resolução de provider + bem-sucedidos, chamada de rede real tentada contra loopback fechado + (nunca saiu da máquina), falha real, retry com backoff de 30s, + dead-letter exatamente na tentativa configurada; `processAnalysisJob` + com transcrição semeada manualmente seguiu o mesmo caminho até a + falha de rede esperada +- [ ] **Nunca exercitado**: chamada de rede real contra OpenAI/Anthropic + (mesma restrição de rede desde o Provider Layer); encadeamento + automático TRANSCRIPTION→ANALYSIS a partir de uma transcrição + *bem-sucedida* (só testável a partir de uma transcrição semeada + manualmente, já que nenhuma chamada real completa sem rede); + convenção de canal 0=cliente/1=agente contra áudio real + distinguível (só tons sintéticos) +- [ ] Scorecards/QA (secao 117-118) e Usage Metering completo/reports de + custo (secao 124) — PHASE 21 + +## PHASE 21+ — ver `agente.md` seções 117 em diante (Scorecards, Usage +Metering, Billing, Frontend, Security, Tests) --- diff --git a/apps/ai-worker/Dockerfile b/apps/ai-worker/Dockerfile new file mode 100644 index 0000000..f1e2eaa --- /dev/null +++ b/apps/ai-worker/Dockerfile @@ -0,0 +1,28 @@ +# syntax=docker/dockerfile:1.7 +# +# Mesmo padrão de apps/predictive-dialer/Dockerfile e +# apps/freeswitch-events/Dockerfile: roda via `tsx` direto, build a partir +# da raiz do monorepo — ver docs/AI_PIPELINE.md. +FROM node:22-slim + +RUN corepack enable && corepack prepare pnpm@11.24.0 --activate + +WORKDIR /repo + +COPY pnpm-workspace.yaml package.json pnpm-lock.yaml tsconfig.base.json ./ +COPY packages/types packages/types +COPY packages/shared packages/shared +COPY packages/database packages/database +COPY packages/storage packages/storage +COPY packages/ai packages/ai +COPY apps/ai-worker apps/ai-worker + +RUN pnpm install --frozen-lockfile --filter @b2bcall/ai-worker... + +# `prisma generate` só precisa do schema, não de uma conexão real. +ENV DATABASE_URL="postgresql://placeholder:placeholder@localhost:5432/placeholder" +RUN pnpm --filter @b2bcall/database exec prisma generate + +WORKDIR /repo/apps/ai-worker + +CMD ["pnpm", "exec", "tsx", "src/main.ts"] diff --git a/apps/ai-worker/package.json b/apps/ai-worker/package.json new file mode 100644 index 0000000..1604b9c --- /dev/null +++ b/apps/ai-worker/package.json @@ -0,0 +1,22 @@ +{ + "name": "@b2bcall/ai-worker", + "version": "0.0.1", + "private": true, + "scripts": { + "dev": "tsx watch src/main.ts", + "build": "tsc -p tsconfig.json", + "start": "node dist/main.js", + "typecheck": "tsc --noEmit" + }, + "dependencies": { + "@b2bcall/ai": "workspace:*", + "@b2bcall/database": "workspace:*", + "@b2bcall/shared": "workspace:*", + "@b2bcall/storage": "workspace:*" + }, + "devDependencies": { + "@types/node": "^22.0.0", + "tsx": "^4.23.12", + "typescript": "^5.7.0" + } +} diff --git a/apps/ai-worker/src/claim.ts b/apps/ai-worker/src/claim.ts new file mode 100644 index 0000000..4257591 --- /dev/null +++ b/apps/ai-worker/src/claim.ts @@ -0,0 +1,41 @@ +import type { Prisma } from "@b2bcall/database"; + +export interface ClaimedJob { + id: string; + callId: string; + type: "TRANSCRIPTION" | "ANALYSIS" | "REANALYSIS"; + attemptCount: number; + maxAttempts: number; +} + +/** + * Reserva atômica de AIJob pendentes (mesmo padrão de + * apps/predictive-dialer/src/lead-reservation.ts § `FOR UPDATE SKIP + * LOCKED`): dois workers concorrentes nunca processam o mesmo job. Precisa + * rodar dentro do MESMO `withTenantContext` transaction — o lock de linha + * só vale até o commit/rollback. + */ +export async function claimPendingJobs( + tx: Prisma.TransactionClient, + tenantId: string, + limit: number, +): Promise { + const rows = await tx.$queryRaw` + SELECT id, call_id AS "callId", type, attempt_count AS "attemptCount", max_attempts AS "maxAttempts" + FROM ai_jobs + WHERE tenant_id = ${tenantId}::uuid + AND status IN ('PENDING', 'RETRYING') + AND scheduled_at <= now() + ORDER BY scheduled_at ASC + LIMIT ${limit} + FOR UPDATE SKIP LOCKED + `; + if (rows.length === 0) return []; + + await tx.aIJob.updateMany({ + where: { id: { in: rows.map((r) => r.id) } }, + data: { status: "PROCESSING" }, + }); + + return rows; +} diff --git a/apps/ai-worker/src/job-outcome.ts b/apps/ai-worker/src/job-outcome.ts new file mode 100644 index 0000000..6c25da7 --- /dev/null +++ b/apps/ai-worker/src/job-outcome.ts @@ -0,0 +1,50 @@ +import { getPrismaClient, withTenantContext } from "@b2bcall/database"; +import { computeNextScheduledAt, isDeadLetter } from "@b2bcall/ai"; +import { createLogger } from "@b2bcall/shared"; + +const logger = createLogger("b2bcall-ai-worker"); + +export async function markJobCompleted(tenantId: string, jobId: string): Promise { + const prisma = getPrismaClient(); + await withTenantContext(prisma, tenantId, (tx) => + tx.aIJob.update({ where: { id: jobId }, data: { status: "COMPLETED", completedAt: new Date() } }), + ); +} + +/** + * Retry/dead-letter (agente.md secao 108): nunca retry infinito. + * `attemptCount` já foi incrementado ANTES de chamar isto — quem processa + * o job soma 1 na entrada, chama o handler, e só chega aqui se o handler + * lançou. + */ +export async function markJobFailed( + tenantId: string, + jobId: string, + attemptCount: number, + maxAttempts: number, + error: unknown, +): Promise { + const prisma = getPrismaClient(); + const message = error instanceof Error ? error.message : String(error); + const deadLetter = isDeadLetter(attemptCount, maxAttempts); + + await withTenantContext(prisma, tenantId, (tx) => + tx.aIJob.update({ + where: { id: jobId }, + data: deadLetter + ? { status: "FAILED", lastError: message, attemptCount } + : { + status: "RETRYING", + lastError: message, + attemptCount, + scheduledAt: computeNextScheduledAt(attemptCount), + }, + }), + ); + + if (deadLetter) { + logger.error("job de IA foi pra dead-letter (tentativas esgotadas)", { jobId, attemptCount, error: message }); + } else { + logger.warn("job de IA falhou, agendado retry com backoff", { jobId, attemptCount, error: message }); + } +} diff --git a/apps/ai-worker/src/main.ts b/apps/ai-worker/src/main.ts new file mode 100644 index 0000000..6e02435 --- /dev/null +++ b/apps/ai-worker/src/main.ts @@ -0,0 +1,81 @@ +import { getPrismaClient, withTenantContext } from "@b2bcall/database"; +import { createLogger } from "@b2bcall/shared"; +import { claimPendingJobs, type ClaimedJob } from "./claim"; +import { processTranscriptionJob } from "./process-transcription"; +import { processAnalysisJob } from "./process-analysis"; +import { markJobCompleted, markJobFailed } from "./job-outcome"; + +const logger = createLogger("b2bcall-ai-worker"); + +const TICK_INTERVAL_MS = Number(process.env.AI_WORKER_TICK_INTERVAL_MS ?? 5000); +const JOBS_PER_TENANT_PER_TICK = Number(process.env.AI_WORKER_JOBS_PER_TICK ?? 5); + +async function processJob(tenantId: string, job: ClaimedJob): Promise { + const attemptCount = job.attemptCount + 1; + try { + if (job.type === "TRANSCRIPTION") { + await processTranscriptionJob(tenantId, job.callId); + } else if (job.type === "ANALYSIS" || job.type === "REANALYSIS") { + await processAnalysisJob(tenantId, job.callId); + } else { + throw new Error(`Tipo de AIJob desconhecido: ${job.type}`); + } + await markJobCompleted(tenantId, job.id); + } catch (err) { + await markJobFailed(tenantId, job.id, attemptCount, job.maxAttempts, err); + } +} + +/** + * Um tick = uma passada por todos os tenants ativos, reservando até + * `JOBS_PER_TENANT_PER_TICK` jobs pendentes de cada um (secao 106-108: + * "pipeline sempre assíncrono, nunca bloqueia a chamada esperando IA"). + * Jobs de tenants diferentes processam em paralelo; dentro de um mesmo + * tenant, sequencial (evita disparar muitas chamadas simultâneas contra o + * mesmo provider/credencial). + */ +async function runTick(): Promise { + const prisma = getPrismaClient(); + const tenants = await prisma.tenant.findMany({ where: { status: "ACTIVE" } }); + + await Promise.all( + tenants.map(async (tenant) => { + const claimed = await withTenantContext(prisma, tenant.id, (tx) => + claimPendingJobs(tx, tenant.id, JOBS_PER_TENANT_PER_TICK), + ); + for (const job of claimed) { + await processJob(tenant.id, job); + } + }), + ); +} + +async function main(): Promise { + let stopped = false; + + const tick = async () => { + if (stopped) return; + try { + await runTick(); + } catch (err) { + logger.error("falha no tick do ai-worker", { error: String(err) }); + } + if (!stopped) setTimeout(tick, TICK_INTERVAL_MS); + }; + setTimeout(tick, TICK_INTERVAL_MS); + + logger.info("b2bcall-ai-worker iniciado", { tickIntervalMs: TICK_INTERVAL_MS }); + + const shutdown = () => { + stopped = true; + logger.info("encerrando b2bcall-ai-worker"); + process.exit(0); + }; + process.on("SIGTERM", shutdown); + process.on("SIGINT", shutdown); +} + +main().catch((err) => { + logger.error("falha fatal ao iniciar b2bcall-ai-worker", { error: String(err) }); + process.exit(1); +}); diff --git a/apps/ai-worker/src/privacy-lookup.ts b/apps/ai-worker/src/privacy-lookup.ts new file mode 100644 index 0000000..e6b1c5d --- /dev/null +++ b/apps/ai-worker/src/privacy-lookup.ts @@ -0,0 +1,37 @@ +import type { Prisma, Tenant } from "@b2bcall/database"; +import { resolvePrivacyLevel, type AIPrivacyLevelName } from "@b2bcall/ai"; + +/** + * Mesma cascata Tenant > Queue > Campaign usada em + * apps/freeswitch-events/src/ai-trigger.ts pra decidir o job de + * TRANSCRIPTION — reconsultada aqui pra decidir se, depois de transcrever, + * a chamada também pode virar um job de ANALYSIS. Recalculada (não lida de + * um valor salvo) porque a config pode ter mudado entre os dois jobs. + */ +export async function resolveEffectivePrivacy( + tx: Prisma.TransactionClient, + tenant: Tenant, + queueId: string | null, + campaignId: string | null, +): Promise { + let queueLevel: AIPrivacyLevelName | null = null; + let campaignTranscriptionEnabled = false; + let campaignAnalysisEnabled = false; + + if (queueId) { + const queue = await tx.queue.findFirst({ where: { id: queueId } }); + queueLevel = (queue?.aiPrivacyLevel as AIPrivacyLevelName | null) ?? null; + } + if (campaignId) { + const campaign = await tx.campaign.findFirst({ where: { id: campaignId } }); + campaignTranscriptionEnabled = campaign?.aiTranscriptionEnabled ?? false; + campaignAnalysisEnabled = campaign?.aiAnalysisEnabled ?? false; + } + + return resolvePrivacyLevel({ + tenantLevel: tenant.aiPrivacyLevel as AIPrivacyLevelName, + queueLevel, + campaignTranscriptionEnabled, + campaignAnalysisEnabled, + }); +} diff --git a/apps/ai-worker/src/process-analysis.ts b/apps/ai-worker/src/process-analysis.ts new file mode 100644 index 0000000..323d191 --- /dev/null +++ b/apps/ai-worker/src/process-analysis.ts @@ -0,0 +1,134 @@ +import { getPrismaClient, withTenantContext, type Prisma } from "@b2bcall/database"; +import { + SensitiveDataRedactor, + CALL_ANALYSIS_JSON_SCHEMA, + validateCallAnalysisResult, +} from "@b2bcall/ai"; +import { createLogger } from "@b2bcall/shared"; +import { resolveProviderForCapability } from "./provider-resolution"; + +const logger = createLogger("b2bcall-ai-worker"); +const redactor = new SensitiveDataRedactor(); + +/** + * Resolução de template (agente.md secao 114-116): `Campaign. + * analysisPromptTemplateId` sobrescreve quando setado; senão o primeiro + * `AIPromptTemplate` de purpose=ANALYSIS com versão ativa, preferindo um + * template do próprio tenant sobre o padrão GLOBAL da plataforma. + */ +async function resolvePromptContent( + tx: Prisma.TransactionClient, + tenantId: string, + campaignId: string | null, +): Promise { + if (campaignId) { + const campaign = await tx.campaign.findFirst({ where: { id: campaignId } }); + if (campaign?.analysisPromptTemplateId) { + const template = await tx.aIPromptTemplate.findFirst({ + where: { id: campaign.analysisPromptTemplateId }, + include: { activeVersion: true }, + }); + if (template?.activeVersion) return template.activeVersion.content; + } + } + + const templates = await tx.aIPromptTemplate.findMany({ + where: { purpose: "ANALYSIS" }, + include: { activeVersion: true }, + }); + const tenantSpecific = templates.find((t) => t.tenantId === tenantId && t.activeVersion); + const global = templates.find((t) => t.tenantId === null && t.activeVersion); + return (tenantSpecific ?? global)?.activeVersion?.content ?? null; +} + +export async function processAnalysisJob(tenantId: string, callId: string): Promise { + const prisma = getPrismaClient(); + + const { call, transcription } = await withTenantContext(prisma, tenantId, async (tx) => { + const call = await tx.call.findUniqueOrThrow({ where: { id: callId } }); + const transcription = await tx.callTranscription.findFirst({ + where: { callId, status: "COMPLETED" }, + orderBy: { createdAt: "desc" }, + }); + return { call, transcription }; + }); + + if (!transcription?.text) { + throw new Error(`Nenhuma transcricao concluida encontrada pra chamada ${callId}`); + } + + const promptContent = await withTenantContext(prisma, tenantId, (tx) => + resolvePromptContent(tx, tenantId, call.campaignId), + ); + if (!promptContent) { + throw new Error("Nenhum AIPromptTemplate (purpose=ANALYSIS) com versao ativa configurado pra este tenant"); + } + + const resolved = await withTenantContext(prisma, tenantId, (tx) => + resolveProviderForCapability(tx, tenantId, "STRUCTURED_OUTPUT"), + ); + if (!resolved) { + throw new Error("Nenhum provider/modelo com capability STRUCTURED_OUTPUT habilitado pra este tenant"); + } + if (!resolved.instance.analyze) { + throw new Error(`Provider ${resolved.providerId} nao implementa analyze()`); + } + + // Secao 123: nunca manda texto sem passar pelo redactor antes de sair + // pra um provider terceiro, independente do nivel de privacidade (que só + // controla SE roda, não O QUE é enviado quando roda). + const redactedText = redactor.redact(transcription.text); + + const result = await resolved.instance.analyze({ + transcriptText: redactedText, + promptContent, + jsonSchema: CALL_ANALYSIS_JSON_SCHEMA, + }); + + const validated = validateCallAnalysisResult(result.data); + + await withTenantContext(prisma, tenantId, async (tx) => { + await tx.callAIAnalysis.create({ + data: { + tenantId, + callId, + providerId: resolved.providerId, + model: resolved.externalModelId, + summary: validated.summary, + customerIntent: validated.customerIntent, + outcome: validated.outcome, + sentiment: validated.sentiment, + topics: validated.topics, + keywords: validated.keywords, + objections: validated.objections, + questions: validated.questions, + actionItems: validated.actionItems, + complianceFlags: validated.complianceFlags, + riskFlags: validated.riskFlags, + qualityScore: validated.qualityScore, + agentScore: validated.agentScore, + customerSentimentScore: validated.customerSentimentScore, + salesOpportunity: validated.salesOpportunity, + nextBestAction: validated.nextBestAction, + }, + }); + + const usageRows = [ + { type: "AI_ANALYSIS_REQUEST" as const, quantity: 1 }, + ...(result.inputTokens ? [{ type: "AI_INPUT_TOKENS" as const, quantity: result.inputTokens }] : []), + ...(result.outputTokens ? [{ type: "AI_OUTPUT_TOKENS" as const, quantity: result.outputTokens }] : []), + ]; + await tx.aIUsageRecord.createMany({ + data: usageRows.map((row) => ({ + tenantId, + callId, + type: row.type, + quantity: row.quantity, + providerId: resolved.providerId, + model: resolved.externalModelId, + })), + }); + }); + + logger.info("analise de IA concluida", { callId, qualityScore: validated.qualityScore }); +} diff --git a/apps/ai-worker/src/process-transcription.ts b/apps/ai-worker/src/process-transcription.ts new file mode 100644 index 0000000..f399bc7 --- /dev/null +++ b/apps/ai-worker/src/process-transcription.ts @@ -0,0 +1,139 @@ +import { join } from "node:path"; +import { getPrismaClient, withTenantContext } from "@b2bcall/database"; +import { getObjectStorageProvider } from "@b2bcall/storage"; +import { splitStereoWav, allowsAnalysis } from "@b2bcall/ai"; +import { createLogger } from "@b2bcall/shared"; +import { createJobWorkDir, cleanupJobWorkDir, streamToFile } from "./temp-files"; +import { resolveProviderForCapability } from "./provider-resolution"; +import { resolveEffectivePrivacy } from "./privacy-lookup"; +import { writeFile } from "node:fs/promises"; + +const logger = createLogger("b2bcall-ai-worker"); + +/** + * Convenção de canal (agente.md secao 111, ver packages/ai/src/ + * wav-stereo-split.ts): canal 0 = perna onde `record_session` foi chamado + * (member/"cliente" no discador), canal 1 = perna bridgeada (agente). Não + * confirmada contra áudio real distinguível nesta sessão. + */ +export async function processTranscriptionJob(tenantId: string, callId: string): Promise { + const prisma = getPrismaClient(); + const workDir = await createJobWorkDir(); + + try { + const { call, tenant, recording } = await withTenantContext(prisma, tenantId, async (tx) => { + const call = await tx.call.findUniqueOrThrow({ where: { id: callId } }); + const tenant = await tx.tenant.findUniqueOrThrow({ where: { id: tenantId } }); + const recording = await tx.recording.findUnique({ where: { callId } }); + return { call, tenant, recording }; + }); + + if (!recording || recording.status !== "AVAILABLE") { + throw new Error(`Gravacao indisponivel pra chamada ${callId} (status: ${recording?.status ?? "inexistente"})`); + } + + const storage = getObjectStorageProvider(); + const stereoPath = join(workDir, "stereo.wav"); + const stream = await storage.getObjectStream(recording.objectKey); + const stereoBuffer = await streamToFile(stream, stereoPath); + + const { channel0, channel1, sampleRate } = splitStereoWav(stereoBuffer); + const ch0Path = join(workDir, "channel0-customer.wav"); + const ch1Path = join(workDir, "channel1-agent.wav"); + await writeFile(ch0Path, channel0); + await writeFile(ch1Path, channel1); + + const resolved = await withTenantContext(prisma, tenantId, (tx) => + resolveProviderForCapability(tx, tenantId, "TRANSCRIPTION"), + ); + if (!resolved) { + throw new Error("Nenhum provider/modelo com capability TRANSCRIPTION habilitado pra este tenant"); + } + if (!resolved.instance.transcribe) { + throw new Error(`Provider ${resolved.providerId} nao implementa transcribe()`); + } + + const [customerResult, agentResult] = await Promise.all([ + resolved.instance.transcribe({ audioFilePath: ch0Path }), + resolved.instance.transcribe({ audioFilePath: ch1Path }), + ]); + + const combinedText = [ + customerResult.text ? `[CUSTOMER]\n${customerResult.text}` : null, + agentResult.text ? `[AGENT]\n${agentResult.text}` : null, + ] + .filter(Boolean) + .join("\n\n"); + + const durationSeconds = + (customerResult.durationSeconds ?? 0) || (agentResult.durationSeconds ?? 0) || recording.durationSeconds || 0; + + await withTenantContext(prisma, tenantId, async (tx) => { + const transcription = await tx.callTranscription.create({ + data: { + tenantId, + callId, + providerId: resolved.providerId, + model: resolved.externalModelId, + language: customerResult.language ?? agentResult.language, + text: combinedText, + status: "COMPLETED", + durationSeconds, + providerRequestId: [customerResult.providerRequestId, agentResult.providerRequestId] + .filter(Boolean) + .join(","), + inputUsage: (customerResult.inputUsage ?? 0) + (agentResult.inputUsage ?? 0), + outputUsage: (customerResult.outputUsage ?? 0) + (agentResult.outputUsage ?? 0), + }, + }); + + const segmentsFor = (result: typeof customerResult, speaker: "CUSTOMER" | "AGENT") => + (result.segments && result.segments.length > 0 + ? result.segments + : result.text + ? [{ startMs: 0, endMs: (result.durationSeconds ?? 0) * 1000, text: result.text }] + : [] + ).map((seg) => ({ + tenantId, + transcriptionId: transcription.id, + speaker, + startMs: seg.startMs, + endMs: seg.endMs, + text: seg.text, + confidence: seg.confidence, + })); + + const segments = [...segmentsFor(customerResult, "CUSTOMER"), ...segmentsFor(agentResult, "AGENT")]; + if (segments.length > 0) { + await tx.callTranscriptSegment.createMany({ data: segments }); + } + + if (durationSeconds > 0) { + await tx.aIUsageRecord.create({ + data: { + tenantId, + callId, + type: "AI_TRANSCRIPTION_SECONDS", + quantity: durationSeconds, + providerId: resolved.providerId, + model: resolved.externalModelId, + }, + }); + } + }); + + logger.info("transcricao concluida", { callId, durationSeconds, sampleRate }); + + const effectiveLevel = await withTenantContext(prisma, tenantId, (tx) => + resolveEffectivePrivacy(tx, tenant, call.queueId, call.campaignId), + ); + if (allowsAnalysis(effectiveLevel)) { + await withTenantContext(prisma, tenantId, (tx) => + tx.aIJob.create({ data: { tenantId, callId, type: "ANALYSIS" } }), + ); + logger.info("job de analise de IA encadeado", { callId }); + } + } finally { + await cleanupJobWorkDir(workDir); + } +} diff --git a/apps/ai-worker/src/provider-resolution.ts b/apps/ai-worker/src/provider-resolution.ts new file mode 100644 index 0000000..aafcb9b --- /dev/null +++ b/apps/ai-worker/src/provider-resolution.ts @@ -0,0 +1,56 @@ +import type { Prisma, AICapability } from "@b2bcall/database"; +import { decryptSecret } from "@b2bcall/shared"; +import { createAIProvider, type AIProvider } from "@b2bcall/ai"; + +export interface ResolvedProvider { + providerId: string; + modelId: string; + externalModelId: string; + instance: AIProvider; +} + +/** + * Resolução de "qual provider/modelo usar pra este job" — agente.md não + * especifica um algoritmo de seleção (só que BYOK deve poder coexistir com + * GLOBAL, secao 100-101). Política adotada, documentada em + * docs/AI_PIPELINE.md: prefere BYOK do próprio tenant (custo e + * credenciais do próprio tenant) sobre o provider GLOBAL da plataforma; + * dentro de cada grupo, o primeiro provider habilitado com um AIModel + * habilitado que tenha a capability pedida, por `createdAt` (determinístico, + * sem heurística de custo/qualidade ainda). Revisar se um dia precisar de + * uma política mais rica (ex.: menor custo, fallback em cadeia). + */ +export async function resolveProviderForCapability( + tx: Prisma.TransactionClient, + tenantId: string, + capability: AICapability, +): Promise { + const candidates = await tx.aIModel.findMany({ + where: { + enabled: true, + capabilities: { has: capability }, + provider: { enabled: true }, + }, + include: { provider: true }, + orderBy: { createdAt: "asc" }, + }); + if (candidates.length === 0) return null; + + const tenantOwned = candidates.filter((c) => c.provider.tenantId === tenantId); + const chosen = tenantOwned[0] ?? candidates.find((c) => c.provider.tenantId === null); + if (!chosen) return null; + + const credentials = { + apiKey: decryptSecret(chosen.provider.encryptedApiKey), + baseUrl: chosen.provider.baseUrl ?? undefined, + organization: chosen.provider.organization ?? undefined, + project: chosen.provider.project ?? undefined, + }; + + return { + providerId: chosen.provider.id, + modelId: chosen.id, + externalModelId: chosen.externalModelId, + instance: createAIProvider(chosen.provider.providerType, credentials), + }; +} diff --git a/apps/ai-worker/src/temp-files.ts b/apps/ai-worker/src/temp-files.ts new file mode 100644 index 0000000..5aa21f2 --- /dev/null +++ b/apps/ai-worker/src/temp-files.ts @@ -0,0 +1,30 @@ +import { randomUUID } from "node:crypto"; +import { tmpdir } from "node:os"; +import { join } from "node:path"; +import { mkdir, rm, writeFile } from "node:fs/promises"; +import type { Readable } from "node:stream"; + +const WORK_ROOT = join(tmpdir(), "b2bcall-ai-worker"); + +/** Um diretório de escopo por job — tudo dentro dele some junto no + * `cleanup`, nunca deixa arquivo órfão em `/tmp` se o processamento + * lançar no meio. */ +export async function createJobWorkDir(): Promise { + const dir = join(WORK_ROOT, randomUUID()); + await mkdir(dir, { recursive: true }); + return dir; +} + +export async function cleanupJobWorkDir(dir: string): Promise { + await rm(dir, { recursive: true, force: true }); +} + +export async function streamToFile(stream: Readable, destPath: string): Promise { + const chunks: Buffer[] = []; + for await (const chunk of stream) { + chunks.push(Buffer.isBuffer(chunk) ? chunk : Buffer.from(chunk)); + } + const buffer = Buffer.concat(chunks); + await writeFile(destPath, buffer); + return buffer; +} diff --git a/apps/ai-worker/tsconfig.json b/apps/ai-worker/tsconfig.json new file mode 100644 index 0000000..3fbff51 --- /dev/null +++ b/apps/ai-worker/tsconfig.json @@ -0,0 +1,9 @@ +{ + "extends": "../../tsconfig.base.json", + "compilerOptions": { + "outDir": "dist", + "rootDir": "src", + "types": ["node"] + }, + "include": ["src"] +} diff --git a/apps/api/src/ai/ai-prompts.controller.ts b/apps/api/src/ai/ai-prompts.controller.ts new file mode 100644 index 0000000..26d54c8 --- /dev/null +++ b/apps/api/src/ai/ai-prompts.controller.ts @@ -0,0 +1,153 @@ +import { Body, Controller, ForbiddenException, Get, NotFoundException, Param, Post, UseGuards } from "@nestjs/common"; +import { getPrismaClient, withTenantContext } from "@b2bcall/database"; +import { recordAuditEvent, isPlatformUser, type AccessTokenClaims } from "@b2bcall/auth"; +import { JwtAuthGuard } from "../common/guards/jwt-auth.guard"; +import { PermissionGuard } from "../common/guards/permission.guard"; +import { RequirePermission } from "../common/decorators/require-permission.decorator"; +import { CurrentUser } from "../common/decorators/current-user.decorator"; +import { CreateAIPromptTemplateDto } from "./dto/create-ai-prompt-template.dto"; +import { CreateAIPromptVersionDto } from "./dto/create-ai-prompt-version.dto"; + +/** + * Templates/versões de prompt (agente.md secao 114-116). Mesma RLS híbrida + * dos providers/models de IA: `tenantId null` = template padrão da + * plataforma, visível de qualquer tenant, só platform admin + * cria/gerencia; `tenantId` setado = só o próprio tenant vê. Versões nunca + * são editadas in-place — cada mudança de conteúdo cria uma linha nova em + * `AIPromptVersion`, `activeVersionId` aponta pra qual está em uso. + */ +@UseGuards(JwtAuthGuard, PermissionGuard) +@Controller("ai/prompt-templates") +export class AIPromptsController { + @RequirePermission("ai.manage") + @Post() + async create(@CurrentUser() user: AccessTokenClaims, @Body() dto: CreateAIPromptTemplateDto) { + const prisma = getPrismaClient(); + const tenantId = user.tenantId!; + const scope = dto.scope ?? "TENANT"; + + if (scope === "GLOBAL") { + const isPlatform = await isPlatformUser(user.sub); + if (!isPlatform) { + throw new ForbiddenException("So' um usuario com role de plataforma pode cadastrar um template GLOBAL"); + } + } + const ownerTenantId = scope === "GLOBAL" ? null : tenantId; + + const template = await withTenantContext(prisma, tenantId, async (tx) => { + const created = await tx.aIPromptTemplate.create({ + data: { tenantId: ownerTenantId, purpose: dto.purpose, name: dto.name }, + }); + if (!dto.content) return created; + + const version = await tx.aIPromptVersion.create({ + data: { tenantId: ownerTenantId, templateId: created.id, version: 1, content: dto.content }, + }); + return tx.aIPromptTemplate.update({ + where: { id: created.id }, + data: { activeVersionId: version.id }, + }); + }); + + await recordAuditEvent(prisma, { + action: "AI_PROMPT_TEMPLATE_CREATE", + tenantId: ownerTenantId, + userId: user.sub, + entityType: "ai_prompt_template", + entityId: template.id, + after: { name: template.name, purpose: template.purpose }, + }); + + return template; + } + + @RequirePermission("ai.view") + @Get() + async list(@CurrentUser() user: AccessTokenClaims) { + const prisma = getPrismaClient(); + const tenantId = user.tenantId!; + return withTenantContext(prisma, tenantId, (tx) => + tx.aIPromptTemplate.findMany({ + include: { activeVersion: true }, + orderBy: { name: "asc" }, + }), + ); + } + + @RequirePermission("ai.manage") + @Post(":id/versions") + async createVersion( + @CurrentUser() user: AccessTokenClaims, + @Param("id") id: string, + @Body() dto: CreateAIPromptVersionDto, + ) { + const prisma = getPrismaClient(); + const tenantId = user.tenantId!; + + const version = await withTenantContext(prisma, tenantId, async (tx) => { + const template = await tx.aIPromptTemplate.findFirst({ where: { id } }); + if (!template) return null; + if (template.tenantId === null) { + const isPlatform = await isPlatformUser(user.sub); + if (!isPlatform) { + throw new ForbiddenException("So' um usuario com role de plataforma pode versionar um template GLOBAL"); + } + } + const last = await tx.aIPromptVersion.findFirst({ + where: { templateId: id }, + orderBy: { version: "desc" }, + }); + return tx.aIPromptVersion.create({ + data: { + tenantId: template.tenantId, + templateId: id, + version: (last?.version ?? 0) + 1, + content: dto.content, + }, + }); + }); + if (!version) throw new NotFoundException("Template nao encontrado"); + + await recordAuditEvent(prisma, { + action: "AI_PROMPT_VERSION_CREATE", + tenantId: version.tenantId, + userId: user.sub, + entityType: "ai_prompt_version", + entityId: version.id, + after: { templateId: id, version: version.version }, + }); + + return version; + } + + @RequirePermission("ai.manage") + @Post(":id/versions/:versionId/activate") + async activateVersion( + @CurrentUser() user: AccessTokenClaims, + @Param("id") id: string, + @Param("versionId") versionId: string, + ) { + const prisma = getPrismaClient(); + const tenantId = user.tenantId!; + + const template = await withTenantContext(prisma, tenantId, async (tx) => { + const existing = await tx.aIPromptTemplate.findFirst({ where: { id } }); + if (!existing) return null; + const version = await tx.aIPromptVersion.findFirst({ where: { id: versionId, templateId: id } }); + if (!version) throw new NotFoundException("Versao nao encontrada para este template"); + return tx.aIPromptTemplate.update({ where: { id }, data: { activeVersionId: versionId } }); + }); + if (!template) throw new NotFoundException("Template nao encontrado"); + + await recordAuditEvent(prisma, { + action: "AI_PROMPT_TEMPLATE_ACTIVATE_VERSION", + tenantId: template.tenantId, + userId: user.sub, + entityType: "ai_prompt_template", + entityId: template.id, + after: { activeVersionId: versionId }, + }); + + return template; + } +} diff --git a/apps/api/src/ai/ai.module.ts b/apps/api/src/ai/ai.module.ts index affaa46..a08f5f6 100644 --- a/apps/api/src/ai/ai.module.ts +++ b/apps/api/src/ai/ai.module.ts @@ -1,8 +1,9 @@ import { Module } from "@nestjs/common"; import { AIProvidersController } from "./ai-providers.controller"; import { AIModelsController } from "./ai-models.controller"; +import { AIPromptsController } from "./ai-prompts.controller"; @Module({ - controllers: [AIProvidersController, AIModelsController], + controllers: [AIProvidersController, AIModelsController, AIPromptsController], }) export class AIModule {} diff --git a/apps/api/src/ai/dto/create-ai-prompt-template.dto.ts b/apps/api/src/ai/dto/create-ai-prompt-template.dto.ts new file mode 100644 index 0000000..745c093 --- /dev/null +++ b/apps/api/src/ai/dto/create-ai-prompt-template.dto.ts @@ -0,0 +1,25 @@ +import { IsIn, IsOptional, IsString, MaxLength } from "class-validator"; + +const PURPOSES = ["ANALYSIS", "SCORECARD"]; +const SCOPES = ["GLOBAL", "TENANT"]; + +export class CreateAIPromptTemplateDto { + @IsIn(PURPOSES) + purpose!: "ANALYSIS" | "SCORECARD"; + + @IsString() + @MaxLength(120) + name!: string; + + /** Mesmo padrão de CreateAIProviderDto — GLOBAL exige isPlatformUser, + * checado no controller, nunca confiado daqui. Default TENANT. */ + @IsOptional() + @IsIn(SCOPES) + scope?: "GLOBAL" | "TENANT"; + + /** Se enviado, já cria a v1 com esse conteúdo e a torna a versão ativa — + * evita 2 chamadas pro caso comum de "criar template com um texto". */ + @IsOptional() + @IsString() + content?: string; +} diff --git a/apps/api/src/ai/dto/create-ai-prompt-version.dto.ts b/apps/api/src/ai/dto/create-ai-prompt-version.dto.ts new file mode 100644 index 0000000..bdd11ff --- /dev/null +++ b/apps/api/src/ai/dto/create-ai-prompt-version.dto.ts @@ -0,0 +1,7 @@ +import { IsString, MinLength } from "class-validator"; + +export class CreateAIPromptVersionDto { + @IsString() + @MinLength(1) + content!: string; +} diff --git a/apps/freeswitch-events/Dockerfile b/apps/freeswitch-events/Dockerfile index 9f0a5cc..d191fc3 100644 --- a/apps/freeswitch-events/Dockerfile +++ b/apps/freeswitch-events/Dockerfile @@ -16,6 +16,8 @@ COPY packages/shared packages/shared COPY packages/telephony packages/telephony COPY packages/database packages/database COPY packages/storage packages/storage +COPY packages/entitlements packages/entitlements +COPY packages/ai packages/ai COPY apps/freeswitch-events apps/freeswitch-events RUN pnpm install --frozen-lockfile --filter @b2bcall/freeswitch-events... diff --git a/apps/freeswitch-events/package.json b/apps/freeswitch-events/package.json index cc7f77d..fabae1a 100644 --- a/apps/freeswitch-events/package.json +++ b/apps/freeswitch-events/package.json @@ -9,7 +9,9 @@ "typecheck": "tsc --noEmit" }, "dependencies": { + "@b2bcall/ai": "workspace:*", "@b2bcall/database": "workspace:*", + "@b2bcall/entitlements": "workspace:*", "@b2bcall/shared": "workspace:*", "@b2bcall/storage": "workspace:*", "@b2bcall/telephony": "workspace:*", diff --git a/apps/freeswitch-events/src/ai-trigger.ts b/apps/freeswitch-events/src/ai-trigger.ts new file mode 100644 index 0000000..fba3efb --- /dev/null +++ b/apps/freeswitch-events/src/ai-trigger.ts @@ -0,0 +1,71 @@ +import { getPrismaClient, withTenantContext, type Tenant, type Plan } from "@b2bcall/database"; +import { isFeatureEnabled } from "@b2bcall/entitlements"; +import { resolvePrivacyLevel, allowsTranscription, type AIPrivacyLevelName } from "@b2bcall/ai"; +import { createLogger } from "@b2bcall/shared"; + +const logger = createLogger("b2bcall-fs-events"); + +/** + * Decide se uma gravação recém-disponível deve virar um AIJob de + * transcrição (agente.md secao 104-108: "pipeline sempre assíncrono, + * disparado depois que a gravação existe"). Duas checagens independentes, + * as duas precisam passar: + * + * 1. Entitlement do Plan (`aiEnabled` + `aiTranscriptionEnabled`) — "essa + * conta contratou IA?", nada a ver com o que o tenant CONFIGUROU. + * 2. Cascata de privacidade Tenant > Queue > Campaign (secao 122) — "esse + * tenant/fila/campanha específica AUTORIZOU processar esta chamada?". + * + * Roda só uma vez, no momento em que a gravação é confirmada — não há + * reconsulta depois (se a config mudar depois, não afeta jobs já + * decididos, mesma filosofia de "decisão tomada no momento do evento" já + * usada pro resto do CDR). + */ +export async function enqueueTranscriptionJobIfEligible( + tenantId: string, + callId: string, + tenant: Tenant & { plan: Plan }, + queueId: string | null, + campaignId: string | null, +): Promise { + try { + const [aiEnabled, transcriptionEnabled] = await Promise.all([ + isFeatureEnabled(tenantId, "aiEnabled"), + isFeatureEnabled(tenantId, "aiTranscriptionEnabled"), + ]); + if (!aiEnabled || !transcriptionEnabled) return; + + const prisma = getPrismaClient(); + + let queueLevel: AIPrivacyLevelName | null = null; + let campaignTranscriptionEnabled = false; + let campaignAnalysisEnabled = false; + + await withTenantContext(prisma, tenantId, async (tx) => { + if (queueId) { + const queue = await tx.queue.findFirst({ where: { id: queueId } }); + queueLevel = (queue?.aiPrivacyLevel as AIPrivacyLevelName | null) ?? null; + } + if (campaignId) { + const campaign = await tx.campaign.findFirst({ where: { id: campaignId } }); + campaignTranscriptionEnabled = campaign?.aiTranscriptionEnabled ?? false; + campaignAnalysisEnabled = campaign?.aiAnalysisEnabled ?? false; + } + }); + + const effectiveLevel = resolvePrivacyLevel({ + tenantLevel: tenant.aiPrivacyLevel as AIPrivacyLevelName, + queueLevel, + campaignTranscriptionEnabled, + campaignAnalysisEnabled, + }); + if (!allowsTranscription(effectiveLevel)) return; + + await withTenantContext(prisma, tenantId, (tx) => + tx.aIJob.create({ data: { tenantId, callId, type: "TRANSCRIPTION" } }), + ); + logger.info("job de transcricao de IA enfileirado", { callId, effectiveLevel }); + } catch (err) { + logger.error("falha ao decidir/enfileirar job de IA", { error: String(err), callId }); + } +} diff --git a/apps/freeswitch-events/src/recording.ts b/apps/freeswitch-events/src/recording.ts index 76bd25c..8085ccd 100644 --- a/apps/freeswitch-events/src/recording.ts +++ b/apps/freeswitch-events/src/recording.ts @@ -3,6 +3,7 @@ import { join } from "node:path"; import { getPrismaClient, withTenantContext } from "@b2bcall/database"; import { getObjectStorageProvider, buildRecordingObjectKey } from "@b2bcall/storage"; import { createLogger } from "@b2bcall/shared"; +import { enqueueTranscriptionJobIfEligible } from "./ai-trigger"; const logger = createLogger("b2bcall-fs-events"); @@ -83,6 +84,8 @@ export async function uploadRecordingIfPresent(tenantId: string, callId: string) await rm(spoolPath, { force: true }); logger.info("gravacao enviada pro object storage", { callId, objectKey, sizeBytes }); + + await enqueueTranscriptionJobIfEligible(tenantId, callId, tenant, call.queueId, call.campaignId); } catch (err) { logger.error("falha ao processar gravacao", { error: String(err), callId }); await withTenantContext(prisma, tenantId, (tx) => diff --git a/docker-compose.yml b/docker-compose.yml index bca9685..8172523 100644 --- a/docker-compose.yml +++ b/docker-compose.yml @@ -162,6 +162,31 @@ services: # não precisa do volume montado aqui, só saber o path Docker-interno. RECORDINGS_SPOOL_DIR: /recordings + ai-worker: + build: + context: . + dockerfile: apps/ai-worker/Dockerfile + container_name: b2bcall-ai-worker + restart: unless-stopped + depends_on: + - postgres + environment: + APP_DATABASE_URL: postgresql://${POSTGRES_APP_USER}:${POSTGRES_APP_PASSWORD}@postgres:5432/${POSTGRES_DB}?schema=public + ENCRYPTION_KEY: ${ENCRYPTION_KEY} + # Precisa ler as gravações que fs-events já subiu (agente.md secao + # 104-108) — mesmo par de env vars/bind mount de fs-events, ver + # docs/RECORDING.md e docs/AI_PIPELINE.md. + STORAGE_PROVIDER: ${STORAGE_PROVIDER:-local} + LOCAL_STORAGE_ROOT: /data/object-storage + S3_BUCKET: ${S3_BUCKET:-} + S3_REGION: ${S3_REGION:-} + S3_ENDPOINT: ${S3_ENDPOINT:-} + S3_ACCESS_KEY_ID: ${S3_ACCESS_KEY_ID:-} + S3_SECRET_ACCESS_KEY: ${S3_SECRET_ACCESS_KEY:-} + S3_FORCE_PATH_STYLE: ${S3_FORCE_PATH_STYLE:-} + volumes: + - ./data/object-storage-local:/data/object-storage + secrets: freeswitch_pat: environment: FREESWITCH_PAT diff --git a/docs/AI_PIPELINE.md b/docs/AI_PIPELINE.md new file mode 100644 index 0000000..3e78141 --- /dev/null +++ b/docs/AI_PIPELINE.md @@ -0,0 +1,118 @@ +# AI Pipeline (agente.md secao 104-116) + +Sub-fase B da fase de IA: transforma o Provider Layer (PHASE 19, ver +`docs/AI_PROVIDERS.md`) num pipeline assíncrono de verdade — transcrição e +análise de chamadas, disparado sozinho depois que uma gravação fica +disponível. + +## Onde cada peça mora + +- `packages/ai/src/privacy.ts` — cascata Tenant > Queue > Campaign (secao + 122), pura, sem I/O. +- `packages/ai/src/retry.ts` — backoff exponencial + dead-letter (secao + 108), pura. +- `packages/ai/src/call-analysis-schema.ts` — schema JSON + validação + manual do resultado de análise (secao 112-113). +- `packages/ai/src/wav-stereo-split.ts` — separa um WAV estéreo em dois + WAV mono (canal 0 = perna onde `record_session` foi chamado/"cliente", + canal 1 = perna bridgeada/"agente" — convenção não confirmada contra + áudio real distinguível nesta sessão, só com tons sintéticos). +- `apps/freeswitch-events/src/ai-trigger.ts` — decide, no momento em que + uma gravação fica `AVAILABLE`, se vira um `AIJob(TRANSCRIPTION)`. + Chamado a partir de `recording.ts` logo depois do `Recording.create`. +- `apps/ai-worker/` — serviço novo (Docker, `tsx` direto, mesmo padrão de + `apps/predictive-dialer`): tick a cada `AI_WORKER_TICK_INTERVAL_MS` + (default 5s), reserva até `AI_WORKER_JOBS_PER_TICK` (default 5) jobs + pendentes por tenant com `FOR UPDATE SKIP LOCKED`, processa + TRANSCRIPTION/ANALYSIS, marca COMPLETED ou aplica retry/dead-letter. +- `apps/api/src/ai/ai-prompts.controller.ts` — CRUD de + `AIPromptTemplate`/`AIPromptVersion` (secao 114-116), mesma RLS híbrida + GLOBAL/tenant dos providers/models. + +## Decisão de disparar (ou não) um job + +Duas checagens independentes, as duas precisam passar (`ai-trigger.ts`): + +1. **Entitlement do Plan** (`aiEnabled` + `aiTranscriptionEnabled`) — "essa + conta contratou IA?". Não-lançante via `isFeatureEnabled` (novo em + `packages/entitlements`), porque isto é uma decisão em background, não + uma requisição HTTP que deveria retornar 403. +2. **Cascata de privacidade** Tenant > Queue > Campaign (secao 122) — + "esse tenant/fila/campanha específica autorizou processar ESTA + chamada?". + +O job de ANALYSIS nunca é criado de cara — só depois que o job de +TRANSCRIPTION correspondente completa com sucesso, reconsultando a mesma +cascata (a config pode ter mudado entre os dois). + +## Resolução de provider/modelo + +agente.md não especifica um algoritmo de seleção quando o tenant tem +múltiplos providers/modelos com a mesma capability. Política adotada +(`apps/ai-worker/src/provider-resolution.ts`): prefere BYOK do próprio +tenant sobre o provider GLOBAL da plataforma; dentro de cada grupo, o +primeiro `AIModel` habilitado com a capability pedida, por `createdAt`. +Sem heurística de custo/qualidade ainda — revisar se precisar de algo mais +rico (ex.: fallback em cadeia, menor custo). + +## Resolução de prompt template (análise) + +`Campaign.analysisPromptTemplateId` sobrescreve quando setado; senão o +primeiro `AIPromptTemplate` de `purpose=ANALYSIS` com versão ativa, +preferindo um template do próprio tenant sobre o padrão GLOBAL. Se nada +for encontrado, o job de ANALYSIS falha com uma mensagem clara ("nenhum +template configurado") e segue o retry/dead-letter normal — não é um +crash, é uma configuração pendente do admin do tenant. + +## Redação de dados sensíveis + +O texto da transcrição sempre passa pelo `SensitiveDataRedactor` (secao +123) antes de sair pro `provider.analyze()`, **independente** do nível de +privacidade — o nível de privacidade controla SE a análise roda, nunca O +QUE é enviado quando roda. Defesa em profundidade. + +## Limitação conhecida: Campaign só pode opt-in, nunca opt-out + +`Campaign.aiTranscriptionEnabled`/`aiAnalysisEnabled` são dois booleanos +independentes (existiam desde a fase Campaigns, secao 63), não o mesmo +enum de 3 níveis de `Tenant`/`Queue`. Uma campanha não consegue forçar +`AI_OFF` explicitamente quando o tenant/fila já tem um nível mais +permissivo — só pode ligar mais IA, nunca desligar. Como "opt-in" é a +direção mais segura por padrão, aceitável por agora. + +## O que foi testado de verdade nesta sessão + +Restrição de rede desta sessão continua valendo: só o servidor git é +autorizado, nenhuma chamada real saiu pra OpenAI/Anthropic. Testado ao +vivo, contra o container `b2bcall-ai-worker` rodando de verdade (não +mocks) e Postgres real com RLS: + +- Cascata de privacidade + entitlement, 3 cenários reais: tenant + permissivo → job criado; `tenant.aiPrivacyLevel=AI_OFF` → nenhum job; + privacidade permitindo mas `Plan.aiEnabled=false` → nenhum job (o + entitlement bloqueia mesmo quando a privacidade libera). +- `splitStereoWav` com um buffer sintético: valores extremos + (±32767/-32768), zero, negativo, todos de-interleaved corretamente; + header malformado lança como esperado. +- `retry.ts`/`privacy.ts`/`call-analysis-schema.ts`: backoff exponencial + com teto, dead-letter no limite exato, cascata de privacidade nos 3 + níveis, validação aceitando/rejeitando resultado de análise conforme + schema. +- Pipeline completo ponta a ponta: WAV estéreo sintético real gravado no + object storage local, `Recording` real `AVAILABLE`, `AIJob` real + reservado via `FOR UPDATE SKIP LOCKED` pelo worker rodando em Docker, + download+split+resolução de provider bem-sucedidos, chamada de rede + real tentada contra uma porta loopback fechada (`http://127.0.0.1:1` — + nunca sai da máquina), falha real (`fetch failed`), retry com backoff + de 30s, dead-letter exatamente na 2ª tentativa configurada. +- `processAnalysisJob` com uma transcrição semeada manualmente: resolveu + template, resolveu provider, redigiu o texto, tentou a chamada real, + falhou da mesma forma esperada. + +**Nunca exercitado**: a chamada de rede de verdade contra a API da OpenAI +ou Anthropic (mesma restrição desde o Provider Layer); o encadeamento +automático TRANSCRIPTION→ANALYSIS após uma transcrição *bem-sucedida* (só +dá pra testar a criação do job de ANALYSIS a partir de uma transcrição +semeada manualmente, já que nenhuma transcrição real chega a completar sem +rede); a convenção de canal 0/1 contra áudio real distinguível (só tons +sintéticos). diff --git a/packages/ai/src/call-analysis-schema.ts b/packages/ai/src/call-analysis-schema.ts new file mode 100644 index 0000000..62f0c88 --- /dev/null +++ b/packages/ai/src/call-analysis-schema.ts @@ -0,0 +1,106 @@ +/** + * Schema do resultado estruturado de análise de chamada (agente.md secao + * 112-113: "não usar somente texto livre... validar antes de persistir"). + * + * `CALL_ANALYSIS_JSON_SCHEMA` é o JSON Schema mandado pro provider (via + * `AnalyzeParams.jsonSchema`) pra guiar/restringir a geração. + * `validateCallAnalysisResult` é a validação de verdade, feita no NOSSO + * lado — nunca confia cegamente que o provider respeitou o schema pedido + * (defesa em profundidade), sem depender de uma lib de JSON Schema + * genérica pra um formato que já é fixo e conhecido. + */ +export const CALL_ANALYSIS_JSON_SCHEMA = { + type: "object", + properties: { + summary: { type: "string" }, + customerIntent: { type: "string" }, + outcome: { type: "string" }, + sentiment: { type: "string", enum: ["POSITIVE", "NEUTRAL", "NEGATIVE"] }, + topics: { type: "array", items: { type: "string" } }, + keywords: { type: "array", items: { type: "string" } }, + objections: { type: "array", items: { type: "string" } }, + questions: { type: "array", items: { type: "string" } }, + actionItems: { type: "array", items: { type: "string" } }, + complianceFlags: { type: "array", items: { type: "string" } }, + riskFlags: { type: "array", items: { type: "string" } }, + qualityScore: { type: "integer", minimum: 0, maximum: 100 }, + agentScore: { type: "integer", minimum: 0, maximum: 100 }, + customerSentimentScore: { type: "number", minimum: -1, maximum: 1 }, + salesOpportunity: { type: "boolean" }, + nextBestAction: { type: "string" }, + }, + required: ["summary", "sentiment", "topics", "qualityScore"], + additionalProperties: false, +} as const; + +export interface CallAnalysisResult { + summary: string; + sentiment: string; + topics: string[]; + qualityScore: number; + customerIntent?: string; + outcome?: string; + keywords: string[]; + objections: string[]; + questions: string[]; + actionItems: string[]; + complianceFlags: string[]; + riskFlags: string[]; + agentScore?: number; + customerSentimentScore?: number; + salesOpportunity?: boolean; + nextBestAction?: string; +} + +function isStringArray(value: unknown): value is string[] { + return Array.isArray(value) && value.every((v) => typeof v === "string"); +} + +export class CallAnalysisValidationError extends Error {} + +export function validateCallAnalysisResult(data: unknown): CallAnalysisResult { + if (typeof data !== "object" || data === null) { + throw new CallAnalysisValidationError("Resultado de análise não é um objeto"); + } + const d = data as Record; + + if (typeof d.summary !== "string" || d.summary.trim() === "") { + throw new CallAnalysisValidationError("summary ausente ou inválido"); + } + if (typeof d.sentiment !== "string") { + throw new CallAnalysisValidationError("sentiment ausente ou inválido"); + } + if (!isStringArray(d.topics)) { + throw new CallAnalysisValidationError("topics ausente ou inválido"); + } + if (typeof d.qualityScore !== "number" || d.qualityScore < 0 || d.qualityScore > 100) { + throw new CallAnalysisValidationError("qualityScore ausente ou fora do intervalo 0-100"); + } + + return { + summary: d.summary, + sentiment: d.sentiment, + topics: d.topics, + qualityScore: Math.round(d.qualityScore), + customerIntent: typeof d.customerIntent === "string" ? d.customerIntent : undefined, + outcome: typeof d.outcome === "string" ? d.outcome : undefined, + keywords: isStringArray(d.keywords) ? d.keywords : [], + objections: isStringArray(d.objections) ? d.objections : [], + questions: isStringArray(d.questions) ? d.questions : [], + actionItems: isStringArray(d.actionItems) ? d.actionItems : [], + complianceFlags: isStringArray(d.complianceFlags) ? d.complianceFlags : [], + riskFlags: isStringArray(d.riskFlags) ? d.riskFlags : [], + agentScore: + typeof d.agentScore === "number" && d.agentScore >= 0 && d.agentScore <= 100 + ? Math.round(d.agentScore) + : undefined, + customerSentimentScore: + typeof d.customerSentimentScore === "number" && + d.customerSentimentScore >= -1 && + d.customerSentimentScore <= 1 + ? d.customerSentimentScore + : undefined, + salesOpportunity: typeof d.salesOpportunity === "boolean" ? d.salesOpportunity : undefined, + nextBestAction: typeof d.nextBestAction === "string" ? d.nextBestAction : undefined, + }; +} diff --git a/packages/ai/src/index.ts b/packages/ai/src/index.ts index 783372d..f660b79 100644 --- a/packages/ai/src/index.ts +++ b/packages/ai/src/index.ts @@ -12,3 +12,18 @@ export { OpenAIProvider } from "./openai-provider"; export { AnthropicProvider } from "./anthropic-provider"; export { createAIProvider, SUPPORTED_PROVIDER_TYPES } from "./registry"; export { SensitiveDataRedactor } from "./redactor"; +export { + CALL_ANALYSIS_JSON_SCHEMA, + validateCallAnalysisResult, + CallAnalysisValidationError, + type CallAnalysisResult, +} from "./call-analysis-schema"; +export { + resolvePrivacyLevel, + allowsTranscription, + allowsAnalysis, + type AIPrivacyLevelName, + type ResolvePrivacyParams, +} from "./privacy"; +export { computeBackoffDelayMs, computeNextScheduledAt, isDeadLetter } from "./retry"; +export { splitStereoWav, type StereoSplitResult } from "./wav-stereo-split"; diff --git a/packages/ai/src/privacy.ts b/packages/ai/src/privacy.ts new file mode 100644 index 0000000..c384a19 --- /dev/null +++ b/packages/ai/src/privacy.ts @@ -0,0 +1,38 @@ +/** + * Resolução de privacidade de IA em cascata (agente.md secao 122): + * Campaign > Queue > Tenant (o nível mais específico configurado vence). + * Tenant é o único nível obrigatório (nunca null no banco, default + * AI_OFF) — os outros dois só existem como *override* opcional. + * + * Simplificação conhecida: `Campaign` guarda a intenção como dois + * booleanos independentes (`aiTranscriptionEnabled`/`aiAnalysisEnabled`, + * já existiam desde a fase Campaigns, secao 63) em vez do enum de 3 + * níveis usado em `Tenant`/`Queue` — não há como uma campanha forçar + * AI_OFF explicitamente quando o tenant/fila já tem um nível mais + * permissivo (só pode fazer *opt-in*, nunca *opt-out* explícito). Dado + * que "opt-in" é a direção mais segura por padrão, aceitável por agora; + * documentado em docs/AI_PIPELINE.md. + */ +export type AIPrivacyLevelName = "AI_OFF" | "TRANSCRIPTION_ONLY" | "TRANSCRIPTION_AND_ANALYSIS"; + +export interface ResolvePrivacyParams { + tenantLevel: AIPrivacyLevelName; + queueLevel?: AIPrivacyLevelName | null; + campaignTranscriptionEnabled?: boolean; + campaignAnalysisEnabled?: boolean; +} + +export function resolvePrivacyLevel(params: ResolvePrivacyParams): AIPrivacyLevelName { + if (params.campaignAnalysisEnabled) return "TRANSCRIPTION_AND_ANALYSIS"; + if (params.campaignTranscriptionEnabled) return "TRANSCRIPTION_ONLY"; + if (params.queueLevel) return params.queueLevel; + return params.tenantLevel; +} + +export function allowsTranscription(level: AIPrivacyLevelName): boolean { + return level === "TRANSCRIPTION_ONLY" || level === "TRANSCRIPTION_AND_ANALYSIS"; +} + +export function allowsAnalysis(level: AIPrivacyLevelName): boolean { + return level === "TRANSCRIPTION_AND_ANALYSIS"; +} diff --git a/packages/ai/src/retry.ts b/packages/ai/src/retry.ts new file mode 100644 index 0000000..9dbacce --- /dev/null +++ b/packages/ai/src/retry.ts @@ -0,0 +1,23 @@ +/** + * Retry/backoff dos AI jobs (agente.md secao 108): "retry, exponential + * backoff, dead-letter strategy, limitar tentativas" — nunca retry + * infinito. + */ +const BASE_DELAY_MS = 30_000; // 30s +const MAX_DELAY_MS = 30 * 60_000; // 30min + +/** Backoff exponencial com teto — 2^attempt * base, nunca passa do teto. */ +export function computeBackoffDelayMs(attemptCount: number): number { + const delay = BASE_DELAY_MS * 2 ** Math.max(0, attemptCount - 1); + return Math.min(delay, MAX_DELAY_MS); +} + +export function computeNextScheduledAt(attemptCount: number, now: Date = new Date()): Date { + return new Date(now.getTime() + computeBackoffDelayMs(attemptCount)); +} + +/** Secao 108: "limitar tentativas" — dead-letter (FAILED terminal) quando + * esgotar. */ +export function isDeadLetter(attemptCount: number, maxAttempts: number): boolean { + return attemptCount >= maxAttempts; +} diff --git a/packages/ai/src/wav-stereo-split.ts b/packages/ai/src/wav-stereo-split.ts new file mode 100644 index 0000000..2ce6d44 --- /dev/null +++ b/packages/ai/src/wav-stereo-split.ts @@ -0,0 +1,118 @@ +/** + * Separação de canais de um WAV estéreo PCM (agente.md secao 111: "sempre + * que possível utilizar canais estéreo pra identificar Agent/Customer... + * não confiar cegamente em diarização quando a direção do áudio permite + * identificação melhor"). Em vez de mandar o áudio estéreo misturado pro + * provider de transcrição (que faria uma diarização probabilística), cada + * canal vira um WAV mono separado, transcrito independentemente — o + * speaker de cada segmento fica 100% determinístico pelo canal de origem, + * não por inferência. + * + * Parser mínimo, não uma lib de áudio genérica: só entende exatamente o + * formato que `record_session`/RECORD_STEREO do FreeSWITCH produz (RIFF/ + * WAVE, PCM linear, 16-bit) — o suficiente pro nosso pipeline, sem trazer + * uma dependência de processamento de áudio (ffmpeg) só pra isso. + * + * Convenção assumida (não verificada contra áudio real distinguível nesta + * sessão — as gravações de teste eram silêncio de `null/dummy`, sem como + * confirmar auditivamente qual canal é qual): canal 0 = a perna em que + * `record_session` foi chamado (o member/"cliente" no fluxo do discador), + * canal 1 = a perna bridgeada (o agente). Revisar com uma chamada real + * antes de confiar no mapeamento em produção — ver docs/AI_PIPELINE.md. + */ + +interface WavInfo { + numChannels: number; + sampleRate: number; + bitsPerSample: number; + dataOffset: number; + dataLength: number; +} + +function parseWavHeader(buffer: Buffer): WavInfo { + if (buffer.toString("ascii", 0, 4) !== "RIFF" || buffer.toString("ascii", 8, 12) !== "WAVE") { + throw new Error("Arquivo nao e' um WAV RIFF valido"); + } + + let offset = 12; + let fmt: { numChannels: number; sampleRate: number; bitsPerSample: number } | undefined; + let dataOffset: number | undefined; + let dataLength: number | undefined; + + while (offset + 8 <= buffer.length) { + const chunkId = buffer.toString("ascii", offset, offset + 4); + const chunkSize = buffer.readUInt32LE(offset + 4); + const bodyStart = offset + 8; + + if (chunkId === "fmt ") { + fmt = { + numChannels: buffer.readUInt16LE(bodyStart + 2), + sampleRate: buffer.readUInt32LE(bodyStart + 4), + bitsPerSample: buffer.readUInt16LE(bodyStart + 14), + }; + } else if (chunkId === "data") { + dataOffset = bodyStart; + dataLength = chunkSize; + } + + offset = bodyStart + chunkSize + (chunkSize % 2); // chunks são alinhados em 2 bytes + } + + if (!fmt || dataOffset === undefined || dataLength === undefined) { + throw new Error("WAV sem chunk fmt/data reconhecível"); + } + if (fmt.bitsPerSample !== 16) { + throw new Error(`Só suporta PCM 16-bit (recebido ${fmt.bitsPerSample}-bit)`); + } + + return { ...fmt, dataOffset, dataLength }; +} + +function buildMonoWav(samples: Buffer, sampleRate: number): Buffer { + const header = Buffer.alloc(44); + header.write("RIFF", 0, "ascii"); + header.writeUInt32LE(36 + samples.length, 4); + header.write("WAVE", 8, "ascii"); + header.write("fmt ", 12, "ascii"); + header.writeUInt32LE(16, 16); // fmt chunk size + header.writeUInt16LE(1, 20); // PCM + header.writeUInt16LE(1, 22); // 1 canal + header.writeUInt32LE(sampleRate, 24); + header.writeUInt32LE(sampleRate * 2, 28); // byte rate (16-bit mono) + header.writeUInt16LE(2, 32); // block align + header.writeUInt16LE(16, 34); // bits per sample + header.write("data", 36, "ascii"); + header.writeUInt32LE(samples.length, 40); + return Buffer.concat([header, samples]); +} + +export interface StereoSplitResult { + channel0: Buffer; + channel1: Buffer; + sampleRate: number; +} + +/** Recebe um WAV estéreo 16-bit e retorna dois buffers WAV mono (um por + * canal), prontos pra transcrever cada um separadamente. */ +export function splitStereoWav(input: Buffer): StereoSplitResult { + const info = parseWavHeader(input); + if (info.numChannels !== 2) { + throw new Error(`Esperava WAV estéreo, recebeu ${info.numChannels} canal(is)`); + } + + const frameCount = Math.floor(info.dataLength / 4); // 2 canais * 2 bytes + const left = Buffer.alloc(frameCount * 2); + const right = Buffer.alloc(frameCount * 2); + + for (let i = 0; i < frameCount; i++) { + const frameOffset = info.dataOffset + i * 4; + left.writeInt16LE(input.readInt16LE(frameOffset), i * 2); + right.writeInt16LE(input.readInt16LE(frameOffset + 2), i * 2); + } + + return { + channel0: buildMonoWav(left, info.sampleRate), + channel1: buildMonoWav(right, info.sampleRate), + sampleRate: info.sampleRate, + }; +} diff --git a/packages/entitlements/src/index.ts b/packages/entitlements/src/index.ts index 5560a94..c28d275 100644 --- a/packages/entitlements/src/index.ts +++ b/packages/entitlements/src/index.ts @@ -72,3 +72,11 @@ export async function assertFeatureEnabled(tenantId: string, key: FeatureKey): P throw new FeatureNotEnabledError(key); } } + +/** Versão não-lançante de `assertFeatureEnabled` — pra decisões em + * background (ex.: "devo enfileirar este job de IA?") onde lançar não faz + * sentido, só pular silenciosamente. */ +export async function isFeatureEnabled(tenantId: string, key: FeatureKey): Promise { + const plan = await getPlanForTenant(tenantId); + return Boolean(plan[key]); +} diff --git a/pnpm-lock.yaml b/pnpm-lock.yaml index 6caa731..45b4261 100644 --- a/pnpm-lock.yaml +++ b/pnpm-lock.yaml @@ -12,6 +12,31 @@ importers: specifier: ^5.9.3 version: 5.9.3 + apps/ai-worker: + dependencies: + '@b2bcall/ai': + specifier: workspace:* + version: link:../../packages/ai + '@b2bcall/database': + specifier: workspace:* + version: link:../../packages/database + '@b2bcall/shared': + specifier: workspace:* + version: link:../../packages/shared + '@b2bcall/storage': + specifier: workspace:* + version: link:../../packages/storage + devDependencies: + '@types/node': + specifier: ^22.0.0 + version: 22.20.1 + tsx: + specifier: ^4.23.12 + version: 4.23.12 + typescript: + specifier: ^5.7.0 + version: 5.9.3 + apps/api: dependencies: '@b2bcall/ai': @@ -127,9 +152,15 @@ importers: apps/freeswitch-events: dependencies: + '@b2bcall/ai': + specifier: workspace:* + version: link:../../packages/ai '@b2bcall/database': specifier: workspace:* version: link:../../packages/database + '@b2bcall/entitlements': + specifier: workspace:* + version: link:../../packages/entitlements '@b2bcall/shared': specifier: workspace:* version: link:../../packages/shared