From 219af5f6f539481b326df7142de09dc7ce48b34b Mon Sep 17 00:00:00 2001 From: Claude Date: Tue, 17 Mar 2026 00:56:14 +0000 Subject: [PATCH] =?UTF-8?q?feat(xos):=20Phase=204=20=E2=80=94=20real-time?= =?UTF-8?q?=20Socket.IO,=20SLA=20scheduler,=20business=20hours,=20campaign?= =?UTF-8?q?s?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - server/socket-io.ts: singleton SocketServer shared across all modules - server/routes.ts: wire initSocketIO before chat/community sockets; start XOS scheduler - server/xos/scheduler.ts: SLA breach checker (every 5min) + supervisor stats broadcaster (every 30s) via broadcastXos - server/xos/routes.ts: campaign segmentation (POST /campaigns/segment), CRUD for xos_campaigns - server/whatsapp/service.ts: XOS queue-based business hours enforcement (xosQueueId in AutoReplyConfig), checkXosQueueIsOpen() helper - client/src/hooks/use-xos-socket.ts: useXosSocket hook — subscribes to xos:supervisor.stats and xos:sla.breach Socket.IO events - client/src/pages/XosSupervisor.tsx: integrate useXosSocket for real-time KPI cards; show connection mode (tempo real vs polling) - shared/schema.ts: fix missing xosDevPipelines pgTable header https://claude.ai/code/session_01DinH3VcgbAv1d9MqnNxzdb --- client/src/hooks/use-xos-socket.ts | 103 +++++++++++++++++++++ client/src/pages/XosSupervisor.tsx | 20 +++-- server/routes.ts | 7 ++ server/socket-io.ts | 32 +++++++ server/whatsapp/service.ts | 46 +++++++++- server/xos/routes.ts | 139 +++++++++++++++++++++++++++++ server/xos/scheduler.ts | 119 ++++++++++++++++++++++++ shared/schema.ts | 2 + 8 files changed, 459 insertions(+), 9 deletions(-) create mode 100644 client/src/hooks/use-xos-socket.ts create mode 100644 server/socket-io.ts create mode 100644 server/xos/scheduler.ts diff --git a/client/src/hooks/use-xos-socket.ts b/client/src/hooks/use-xos-socket.ts new file mode 100644 index 0000000..ae10986 --- /dev/null +++ b/client/src/hooks/use-xos-socket.ts @@ -0,0 +1,103 @@ +/** + * useXosSocket — subscribes to XOS real-time events from Socket.IO. + * Usage: + * const { supervisorStats, lastSlaBreachEvent } = useXosSocket(); + */ +import { useEffect, useRef, useState, useCallback } from "react"; +import { io, Socket } from "socket.io-client"; + +export interface SupervisorStats { + openConversations: number; + resolvedToday: number; + urgentTickets: number; + agentsOnline: number; + timestamp: string; +} + +export interface SlaBreachEvent { + protocolId: number; + protocolNumber: string; + tenantId: number; + queueId: number | null; + assignedTo: number | null; + breachedAt: string; +} + +interface XosSocketState { + connected: boolean; + supervisorStats: SupervisorStats | null; + lastSlaBreachEvent: SlaBreachEvent | null; +} + +let _socket: Socket | null = null; +let _refCount = 0; + +function getSocket(): Socket { + if (!_socket) { + _socket = io(window.location.origin, { + path: "/socket.io", + transports: ["websocket", "polling"], + reconnectionAttempts: 10, + reconnectionDelay: 2000, + }); + } + return _socket; +} + +export function useXosSocket() { + const [state, setState] = useState({ + connected: false, + supervisorStats: null, + lastSlaBreachEvent: null, + }); + const socketRef = useRef(null); + + const handleStats = useCallback((data: SupervisorStats) => { + setState((s) => ({ ...s, supervisorStats: data })); + }, []); + + const handleSlaBreach = useCallback((data: SlaBreachEvent) => { + setState((s) => ({ ...s, lastSlaBreachEvent: data })); + }, []); + + const handleConnect = useCallback(() => { + setState((s) => ({ ...s, connected: true })); + }, []); + + const handleDisconnect = useCallback(() => { + setState((s) => ({ ...s, connected: false })); + }, []); + + useEffect(() => { + const socket = getSocket(); + socketRef.current = socket; + _refCount++; + + socket.on("connect", handleConnect); + socket.on("disconnect", handleDisconnect); + socket.on("xos:supervisor.stats", handleStats); + socket.on("xos:sla.breach", handleSlaBreach); + + // Sync initial connection state + if (socket.connected) { + setState((s) => ({ ...s, connected: true })); + } + + return () => { + socket.off("connect", handleConnect); + socket.off("disconnect", handleDisconnect); + socket.off("xos:supervisor.stats", handleStats); + socket.off("xos:sla.breach", handleSlaBreach); + + _refCount--; + // Keep socket alive as long as any component uses it + if (_refCount <= 0 && _socket) { + _socket.disconnect(); + _socket = null; + _refCount = 0; + } + }; + }, [handleConnect, handleDisconnect, handleStats, handleSlaBreach]); + + return state; +} diff --git a/client/src/pages/XosSupervisor.tsx b/client/src/pages/XosSupervisor.tsx index 038ca47..154baf6 100644 --- a/client/src/pages/XosSupervisor.tsx +++ b/client/src/pages/XosSupervisor.tsx @@ -1,5 +1,6 @@ import { useState } from "react"; import { useQuery } from "@tanstack/react-query"; +import { useXosSocket } from "@/hooks/use-xos-socket"; import { Link } from "wouter"; import { BrowserFrame } from "@/components/Browser/BrowserFrame"; import { @@ -71,6 +72,7 @@ export default function XosSupervisor() { const [activeTab, setActiveTab] = useState("overview"); const [queueFilter, setQueueFilter] = useState("all"); const [autoRefresh] = useState(true); + const { connected: socketConnected, supervisorStats: liveStats } = useXosSocket(); const refetchInterval = autoRefresh ? 30000 : false; @@ -143,9 +145,13 @@ export default function XosSupervisor() { ); } + // Prefer real-time socket stats over polling data when available const conv = overview?.conversations; const tick = overview?.tickets; - const totalActive = (conv?.open || 0) + (conv?.pending || 0); + const totalActive = liveStats?.openConversations ?? ((conv?.open || 0) + (conv?.pending || 0)); + const resolvedToday = liveStats?.resolvedToday ?? conv?.resolved_today ?? 0; + const urgentTickets = liveStats?.urgentTickets ?? tick?.urgent_open ?? 0; + const agentsOnline = liveStats?.agentsOnline ?? overview?.agents_active?.length ?? 0; return ( @@ -167,7 +173,10 @@ export default function XosSupervisor() {

Monitor de Supervisor

Atualizado às {overview ? new Date(overview.generated_at).toLocaleTimeString("pt-BR") : "—"} - {autoRefresh && • ao vivo} + {socketConnected + ? ● tempo real + : autoRefresh && ○ polling 30s + }

@@ -205,6 +214,7 @@ export default function XosSupervisor() {
{conv?.unassigned || 0} sem agente + • {conv?.pending || 0} pendentes
@@ -216,7 +226,7 @@ export default function XosSupervisor() {

Resolvidos Hoje

-

{conv?.resolved_today || 0}

+

{resolvedToday}

@@ -231,7 +241,7 @@ export default function XosSupervisor() {

Tickets Urgentes

-

{tick?.urgent_open || 0}

+

{urgentTickets}

@@ -252,7 +262,7 @@ export default function XosSupervisor() {

Agentes Ativos

-

{overview?.agents_active?.length || 0}

+

{agentsOnline}

diff --git a/server/routes.ts b/server/routes.ts index ca0bb24..dc96902 100644 --- a/server/routes.ts +++ b/server/routes.ts @@ -61,6 +61,8 @@ import pipelineRoutes from "./blackboard/pipelineRoutes"; import { startAllAgents } from "./blackboard/agents"; import { loadModuleRoutes } from "./modules/loader"; import graphRoutes from "./graph/routes"; +import { initSocketIO } from "./socket-io"; +import { startXosScheduler } from "./xos/scheduler"; export async function registerRoutes( httpServer: Server, @@ -82,6 +84,8 @@ export async function registerRoutes( registerChatRoutes(app); registerSoeRoutes(app); registerInternalChatRoutes(app); + // Initialize shared Socket.IO singleton (must be before other socket setups) + initSocketIO(httpServer); setupChatSocket(httpServer); setupCommunitySocket(httpServer); registerWhatsappRoutes(app); @@ -144,6 +148,9 @@ export async function registerRoutes( // Iniciar os 6 agentes do Blackboard startAllAgents(); + + // XOS Scheduler: SLA breach checker + supervisor stats broadcaster + startXosScheduler(); // Central de Protocolos (MCP, A2A, AP2, UCP) app.use("/api", protocolsRoutes); diff --git a/server/socket-io.ts b/server/socket-io.ts new file mode 100644 index 0000000..3df09c0 --- /dev/null +++ b/server/socket-io.ts @@ -0,0 +1,32 @@ +/** + * Singleton Socket.IO — instância compartilhada entre todos os módulos. + * Inicializado em routes.ts via initSocketIO(httpServer). + * Qualquer módulo pode importar getIO() para emitir eventos. + */ +import { Server as HttpServer } from "http"; +import { Server as SocketServer, Socket } from "socket.io"; + +let _io: SocketServer | null = null; + +export function initSocketIO(httpServer: HttpServer): SocketServer { + if (_io) return _io; + _io = new SocketServer(httpServer, { + path: "/socket.io", + cors: { origin: "*", methods: ["GET", "POST"] }, + }); + return _io; +} + +export function getIO(): SocketServer { + if (!_io) throw new Error("[socket-io] Not initialized. Call initSocketIO first."); + return _io; +} + +/** Emite um evento XOS para todos os clientes conectados (não bloqueia). */ +export function broadcastXos(event: string, data: unknown) { + try { + getIO().emit(`xos:${event}`, data); + } catch { + // Socket ainda não iniciado ou sem clientes — ignorar silenciosamente + } +} diff --git a/server/whatsapp/service.ts b/server/whatsapp/service.ts index 159ac7f..23f9eb4 100644 --- a/server/whatsapp/service.ts +++ b/server/whatsapp/service.ts @@ -26,6 +26,8 @@ interface AutoReplyConfig { outsideHoursMessage: string; aiEnabled: boolean; maxAutoRepliesPerContact: number; + /** Optional: link to an XOS queue for schedule/out-of-hours config */ + xosQueueId?: number; } interface WhatsAppSession { @@ -130,6 +132,31 @@ Nome do cliente: ${contactName}`; } } + /** Check XOS queue schedule. Returns { isOpen, outOfHoursMessage }. */ + private async checkXosQueueIsOpen(queueId: number): Promise<{ isOpen: boolean; outOfHoursMessage: string | null }> { + try { + const result = await db.execute(sql` + SELECT schedules, out_of_hours_message FROM xos_queues WHERE id = ${queueId} + `); + const queue = ((result as any).rows ?? [])[0]; + if (!queue) return { isOpen: true, outOfHoursMessage: null }; + + const schedules: Array<{ dayOfWeek: number; startTime: string; endTime: string; enabled: boolean }> = queue.schedules || []; + if (!schedules.length) return { isOpen: true, outOfHoursMessage: null }; + + const now = new Date(); + const dayOfWeek = now.getDay(); // 0=Sun, 6=Sat + const currentTime = `${String(now.getHours()).padStart(2, "0")}:${String(now.getMinutes()).padStart(2, "0")}`; + const todaySchedule = schedules.find((s) => s.dayOfWeek === dayOfWeek && s.enabled !== false); + + if (!todaySchedule) return { isOpen: false, outOfHoursMessage: queue.out_of_hours_message }; + const isOpen = currentTime >= todaySchedule.startTime && currentTime < todaySchedule.endTime; + return { isOpen, outOfHoursMessage: isOpen ? null : queue.out_of_hours_message }; + } catch { + return { isOpen: true, outOfHoursMessage: null }; // fail-open: don't block messages if DB query fails + } + } + private async processAutoReply(msg: IncomingMessage, contact: typeof whatsappContacts.$inferSelect): Promise { try { const config = this.getAutoReplyConfig(msg.userId); @@ -137,18 +164,29 @@ Nome do cliente: ${contactName}`; const contactKey = `${msg.userId}_${contact.id}`; const currentCount = this.autoReplyCount.get(contactKey) || 0; - + if (currentCount >= config.maxAutoRepliesPerContact) { return; } - const currentHour = new Date().getHours(); - const isBusinessHours = currentHour >= config.businessHours.start && currentHour < config.businessHours.end; + // Check business hours — prefer XOS queue schedule when configured + let isBusinessHours: boolean; + let outsideHoursReply: string; + + if (config.xosQueueId) { + const { isOpen, outOfHoursMessage } = await this.checkXosQueueIsOpen(config.xosQueueId); + isBusinessHours = isOpen; + outsideHoursReply = outOfHoursMessage || config.outsideHoursMessage; + } else { + const currentHour = new Date().getHours(); + isBusinessHours = currentHour >= config.businessHours.start && currentHour < config.businessHours.end; + outsideHoursReply = config.outsideHoursMessage; + } let replyText: string; if (!isBusinessHours) { - replyText = config.outsideHoursMessage; + replyText = outsideHoursReply; } else if (currentCount === 0) { replyText = config.welcomeMessage; } else if (config.aiEnabled) { diff --git a/server/xos/routes.ts b/server/xos/routes.ts index 5e57e48..68e03e1 100644 --- a/server/xos/routes.ts +++ b/server/xos/routes.ts @@ -1646,4 +1646,143 @@ export async function fireCrmAutomations(eventType: string, payload: Record { + try { + const { tenantId, filters = {}, limit = 500, offset = 0 } = req.body; + const safeLimit = Math.min(Number(limit) || 500, 5000); + const safeOffset = Math.max(Number(offset) || 0, 0); + + // Build SQL fragments dynamically + const clauses: ReturnType[] = [sql`1=1`]; + + if (tenantId) clauses.push(sql`c.tenant_id = ${tenantId}`); + if (filters.type) clauses.push(sql`c.type = ${filters.type}`); + if (filters.city) clauses.push(sql`c.city ILIKE ${"%" + filters.city + "%"}`); + if (filters.state) clauses.push(sql`c.state ILIKE ${"%" + filters.state + "%"}`); + if (filters.assignedTo) clauses.push(sql`c.assigned_to = ${filters.assignedTo}`); + if (filters.hasPhone === true) clauses.push(sql`(c.phone IS NOT NULL AND c.phone != '')`); + if (filters.hasPhone === false) clauses.push(sql`(c.phone IS NULL OR c.phone = '')`); + if (filters.hasWhatsapp === true) clauses.push(sql`(c.whatsapp IS NOT NULL AND c.whatsapp != '')`); + if (filters.hasWhatsapp === false) clauses.push(sql`(c.whatsapp IS NULL OR c.whatsapp = '')`); + if (filters.hasEmail === true) clauses.push(sql`(c.email IS NOT NULL AND c.email != '')`); + if (filters.hasEmail === false) clauses.push(sql`(c.email IS NULL OR c.email = '')`); + if (typeof filters.leadScoreMin === "number") clauses.push(sql`c.lead_score >= ${filters.leadScoreMin}`); + if (typeof filters.leadScoreMax === "number") clauses.push(sql`c.lead_score <= ${filters.leadScoreMax}`); + if (Array.isArray(filters.leadStatus) && filters.leadStatus.length > 0) { + clauses.push(sql`c.lead_status = ANY(${filters.leadStatus})`); + } + if (Array.isArray(filters.source) && filters.source.length > 0) { + clauses.push(sql`c.source = ANY(${filters.source})`); + } + if (Array.isArray(filters.tags) && filters.tags.length > 0) { + clauses.push(sql`c.tags @> ${filters.tags}::text[]`); + } + + const whereSql = sql.join(clauses, sql` AND `); + + const countResult = await db.execute(sql`SELECT COUNT(*) AS total FROM xos_contacts c WHERE ${whereSql}`); + const total = Number(((countResult as any).rows?.[0])?.total ?? 0); + + const dataResult = await db.execute(sql` + SELECT c.id, c.name, c.email, c.phone, c.whatsapp, c.type, c.lead_status, + c.lead_score, c.source, c.tags, c.city, c.state, c.company, + c.assigned_to, c.last_contact_at + FROM xos_contacts c + WHERE ${whereSql} + ORDER BY c.name ASC + LIMIT ${safeLimit} OFFSET ${safeOffset} + `); + + const contacts = (dataResult as any).rows ?? []; + res.json({ total, limit: safeLimit, offset: safeOffset, count: contacts.length, contacts }); + } catch (error: any) { + console.error("Campaign segment error:", error); + res.status(500).json({ error: "Failed to query campaign segment" }); + } +}); + +/** + * GET /api/xos/campaigns — list campaigns for tenant + */ +router.get("/campaigns", async (req: Request, res: Response) => { + try { + const tenantId = (req as any).user?.tenantId; + const result = await db.execute(sql` + SELECT id, name, description, type, status, segment_query, subject, + scheduled_at, started_at, completed_at, stats, created_at, updated_at + FROM xos_campaigns + ${tenantId ? sql`WHERE tenant_id = ${tenantId}` : sql``} + ORDER BY created_at DESC + LIMIT 100 + `); + res.json((result as any).rows ?? []); + } catch (error: any) { + console.error("List campaigns error:", error); + res.status(500).json({ error: "Failed to list campaigns" }); + } +}); + +/** + * POST /api/xos/campaigns — create campaign + */ +router.post("/campaigns", async (req: Request, res: Response) => { + try { + const tenantId = (req as any).user?.tenantId; + const userId = (req as any).user?.id; + const { name, description, type, segmentQuery, content, subject, scheduledAt } = req.body; + if (!name || !type) return res.status(400).json({ error: "name and type are required" }); + + const result = await db.execute(sql` + INSERT INTO xos_campaigns (tenant_id, name, description, type, status, segment_query, content, subject, scheduled_at, created_by) + VALUES ( + ${tenantId || null}, ${name}, ${description || null}, ${type}, 'draft', + ${segmentQuery ? JSON.stringify(segmentQuery) : null}, + ${content || null}, ${subject || null}, ${scheduledAt || null}, ${userId || null} + ) + RETURNING * + `); + res.status(201).json(((result as any).rows ?? [])[0]); + } catch (error: any) { + console.error("Create campaign error:", error); + res.status(500).json({ error: "Failed to create campaign" }); + } +}); + +/** + * PATCH /api/xos/campaigns/:id — update campaign + */ +router.patch("/campaigns/:id", async (req: Request, res: Response) => { + try { + const id = Number(req.params.id); + const { status, name, description, content, subject, scheduledAt, segmentQuery } = req.body; + await db.execute(sql` + UPDATE xos_campaigns SET + status = COALESCE(${status || null}, status), + name = COALESCE(${name || null}, name), + description = COALESCE(${description || null}, description), + content = COALESCE(${content || null}, content), + subject = COALESCE(${subject || null}, subject), + scheduled_at = COALESCE(${scheduledAt || null}, scheduled_at), + segment_query = COALESCE(${segmentQuery ? JSON.stringify(segmentQuery) : null}::jsonb, segment_query), + started_at = CASE WHEN ${status || null} = 'running' AND started_at IS NULL THEN NOW() ELSE started_at END, + completed_at = CASE WHEN ${status || null} IN ('completed','cancelled') AND completed_at IS NULL THEN NOW() ELSE completed_at END, + updated_at = NOW() + WHERE id = ${id} + `); + res.json({ ok: true }); + } catch (error: any) { + console.error("Update campaign error:", error); + res.status(500).json({ error: "Failed to update campaign" }); + } +}); + export default router; diff --git a/server/xos/scheduler.ts b/server/xos/scheduler.ts new file mode 100644 index 0000000..80e35cf --- /dev/null +++ b/server/xos/scheduler.ts @@ -0,0 +1,119 @@ +/** + * XOS Scheduler + * - Every 5 min: check SLA breaches and emit events + * - Every 30 sec: broadcast live supervisor stats via Socket.IO + */ +import { db } from "../db"; +import { sql } from "drizzle-orm"; +import { broadcastXos } from "../socket-io"; + +const SLA_CHECK_INTERVAL_MS = 5 * 60 * 1000; // 5 minutes +const STATS_BROADCAST_INTERVAL_MS = 30 * 1000; // 30 seconds + +/** Mark overdue protocols as SLA-breached and emit event per tenant */ +async function checkSlaBreaches() { + try { + // Mark protocols where sla_deadline has passed and still open/unbreached + const breached = await db.execute(sql` + UPDATE xos_protocols + SET sla_breach = true, updated_at = NOW() + WHERE status = 'open' + AND sla_deadline IS NOT NULL + AND sla_deadline < NOW() + AND sla_breach = false + RETURNING id, tenant_id, protocol_number, contact_id, queue_id, assigned_to + `); + + const rows = (breached as any).rows ?? []; + for (const row of rows) { + broadcastXos("sla.breach", { + protocolId: row.id, + tenantId: row.tenant_id, + protocolNumber: row.protocol_number, + contactId: row.contact_id, + queueId: row.queue_id, + assignedTo: row.assigned_to, + breachedAt: new Date().toISOString(), + }); + + // Forward to automation engine (non-blocking) + try { + const engineHost = process.env.AUTOMATION_ENGINE_HOST || "localhost"; + const enginePort = process.env.AUTOMATION_ENGINE_PORT || "8005"; + await fetch(`http://${engineHost}:${enginePort}/xos/trigger`, { + method: "POST", + headers: { "Content-Type": "application/json" }, + body: JSON.stringify({ + event_type: "crm.sla.breached", + payload: { protocol_id: row.id, protocol_number: row.protocol_number }, + tenant_id: row.tenant_id, + }), + signal: AbortSignal.timeout(5000), + }); + } catch { + // Non-blocking — engine may be offline + } + } + + if (rows.length > 0) { + console.log(`[xos:scheduler] SLA check: ${rows.length} breach(es) detected`); + } + } catch (err) { + console.error("[xos:scheduler] SLA check error:", err); + } +} + +/** Broadcast aggregate supervisor stats to all connected clients */ +async function broadcastSupervisorStats() { + try { + const [convsResult, ticketsResult, agentsResult] = await Promise.all([ + db.execute(sql` + SELECT + COUNT(*) FILTER (WHERE status = 'open') AS open_conversations, + COUNT(*) FILTER (WHERE status = 'resolved' AND updated_at > NOW() - INTERVAL '24h') AS resolved_today + FROM xos_conversations + `), + db.execute(sql` + SELECT + COUNT(*) FILTER (WHERE status IN ('open','in_progress') AND priority = 'urgent') AS urgent_open + FROM xos_tickets + `), + db.execute(sql` + SELECT COUNT(*) FILTER (WHERE is_online = true) AS agents_online + FROM xos_agents + `), + ]); + + const c = ((convsResult as any).rows ?? [])[0] ?? {}; + const t = ((ticketsResult as any).rows ?? [])[0] ?? {}; + const a = ((agentsResult as any).rows ?? [])[0] ?? {}; + + broadcastXos("supervisor.stats", { + openConversations: Number(c.open_conversations ?? 0), + resolvedToday: Number(c.resolved_today ?? 0), + urgentTickets: Number(t.urgent_open ?? 0), + agentsOnline: Number(a.agents_online ?? 0), + timestamp: new Date().toISOString(), + }); + } catch (err) { + // Suppress — tables may not exist in dev + if (process.env.NODE_ENV !== "production") { + console.debug("[xos:scheduler] stats broadcast skipped:", (err as any)?.message); + } + } +} + +export function startXosScheduler() { + // Stagger initial runs slightly to avoid startup congestion + setTimeout(() => { + checkSlaBreaches(); + setInterval(checkSlaBreaches, SLA_CHECK_INTERVAL_MS); + }, 15_000); // first SLA check 15s after boot + + setTimeout(() => { + broadcastSupervisorStats(); + setInterval(broadcastSupervisorStats, STATS_BROADCAST_INTERVAL_MS); + }, 5_000); // first broadcast 5s after boot + + console.log("[xos:scheduler] Started — SLA checks every 5min, stats every 30s"); +} diff --git a/shared/schema.ts b/shared/schema.ts index d3aab9d..66183a0 100644 --- a/shared/schema.ts +++ b/shared/schema.ts @@ -7145,6 +7145,8 @@ export const xosSlaPolicies = pgTable("xos_sla_policies", { export const insertXosSlaPolicySchema = createInsertSchema(xosSlaPolicies).omit({ id: true, createdAt: true, updatedAt: true }); export type XosSlaPolicy = typeof xosSlaPolicies.$inferSelect; export type InsertXosSlaPolicy = z.infer; + +export const xosDevPipelines = pgTable("xos_dev_pipelines", { id: serial("id").primaryKey(), correlationId: text("correlation_id").notNull().default(sql`gen_random_uuid()`), prompt: text("prompt").notNull(),