|
|
@@ -4,6 +4,9 @@ import { config } from '../config.js';
|
|
|
import { logger } from '../logger.js';
|
|
|
import { audit } from '../store/auditLog.js';
|
|
|
import { setMessageStatus } from '../store/idempotencyStore.js';
|
|
|
+import { evaluateRuntimeSkip, getFlowiseSessionId } from '../store/runtimeSettings.js';
|
|
|
+import { RelayError } from '../errors.js';
|
|
|
+import { classifyAttachments } from './attachmentPolicy.js';
|
|
|
import { evaluateSpam } from './spamGate.js';
|
|
|
import { createTicket, hasLocalTicket, isTicketMode } from './ticketService.js';
|
|
|
import type { SupportMessageEvent } from './messageNormalizer.js';
|
|
|
@@ -18,6 +21,7 @@ export type PipelineOutcome =
|
|
|
| { action: 'spam'; reason: string }
|
|
|
| { action: 'reply' }
|
|
|
| { action: 'handoff'; ticketNumber: string }
|
|
|
+ | { action: 'ai_error'; code: string }
|
|
|
| { action: 'no_reply'; reason: string };
|
|
|
|
|
|
/**
|
|
|
@@ -53,6 +57,19 @@ export async function processMessageEvent(
|
|
|
additionalAttributes = { ...additionalAttributes, ...(conv.additional_attributes ?? {}) };
|
|
|
}
|
|
|
|
|
|
+ const runtime = await evaluateRuntimeSkip(event.messageCreatedAt ? new Date(event.messageCreatedAt) : null);
|
|
|
+ if (runtime.decision.skip) {
|
|
|
+ await setMessageStatus(event.source, event.messageId, 'skipped', runtime.decision.reason);
|
|
|
+ await audit({
|
|
|
+ conversationId: event.conversationId,
|
|
|
+ messageId: event.messageId,
|
|
|
+ eventType: `skipped_${runtime.decision.reason}`,
|
|
|
+ summary: `Message stopped by runtime controls: ${runtime.decision.reason}`,
|
|
|
+ meta: { stage: 'pipeline', messageCreatedAt: event.messageCreatedAt },
|
|
|
+ });
|
|
|
+ return { action: 'skipped', reason: runtime.decision.reason };
|
|
|
+ }
|
|
|
+
|
|
|
if (cfg.CHATWOOT_REQUIRE_AGENT_BOT_ASSIGNMENT && conv && !isAssignedToRequiredAgentBot(conv)) {
|
|
|
await setMessageStatus(event.source, event.messageId, 'skipped', 'not_assigned_to_bot');
|
|
|
await audit({
|
|
|
@@ -104,14 +121,74 @@ export async function processMessageEvent(
|
|
|
}
|
|
|
}
|
|
|
|
|
|
+ if (event.formMail) {
|
|
|
+ await audit({
|
|
|
+ conversationId: event.conversationId,
|
|
|
+ messageId: event.messageId,
|
|
|
+ eventType: 'form_mail_detected',
|
|
|
+ summary: 'WWW/contact-form mail detected and mapped to customer sender',
|
|
|
+ meta: { technicalSender: event.formMail.technicalSender, hasPhone: event.formMail.phone !== '' },
|
|
|
+ });
|
|
|
+ try {
|
|
|
+ await chatwoot.setCustomAttributes(event.conversationId, {
|
|
|
+ form_mail: true,
|
|
|
+ form_customer_email: event.formMail.customerEmail,
|
|
|
+ form_source: event.formMail.formSource,
|
|
|
+ });
|
|
|
+ await chatwoot.addLabel(event.conversationId, 'www-form');
|
|
|
+ } catch {
|
|
|
+ logger.warn('Could not persist form-mail attributes (non-fatal)', { conversationId: event.conversationId });
|
|
|
+ }
|
|
|
+ }
|
|
|
+
|
|
|
+ const attachmentVerdict = classifyAttachments(event.attachments, event.content);
|
|
|
+ if (attachmentVerdict.requiresHandoff) {
|
|
|
+ const ticket = await createTicket(event.conversationId, attachmentVerdict.reason, chatwoot);
|
|
|
+ try {
|
|
|
+ await chatwoot.addLabel(event.conversationId, 'attachment');
|
|
|
+ await chatwoot.sendPrivateNote(
|
|
|
+ event.conversationId,
|
|
|
+ `Wiadomość zawiera istotny lub nieobsługiwany załącznik; AI nie analizowała pliku. Ticket: ${ticket.ticketNumber}.`,
|
|
|
+ );
|
|
|
+ } catch {
|
|
|
+ logger.warn('Could not persist attachment handoff note/label (non-fatal)', { conversationId: event.conversationId });
|
|
|
+ }
|
|
|
+ await setMessageStatus(event.source, event.messageId, 'ticket', 'attachment_handoff');
|
|
|
+ await audit({
|
|
|
+ conversationId: event.conversationId,
|
|
|
+ messageId: event.messageId,
|
|
|
+ eventType: 'attachment_handoff',
|
|
|
+ summary: 'Message contains a significant attachment — handed off before Flowise',
|
|
|
+ meta: { significantCount: attachmentVerdict.significant.length },
|
|
|
+ });
|
|
|
+ return { action: 'handoff', ticketNumber: ticket.ticketNumber };
|
|
|
+ }
|
|
|
+
|
|
|
// Snapshot before the LLM turn: the agent may call /tools/new_ticket while
|
|
|
// Flowise is thinking, and afterwards we need to tell "ticketed just now"
|
|
|
// apart from "a ticket row already existed from an earlier, since-reopened
|
|
|
// conversation".
|
|
|
const hadTicketBefore = await hasLocalTicket(event.conversationId);
|
|
|
|
|
|
- const payload = buildFlowisePayload(event);
|
|
|
- const response = await flowise.predict(payload);
|
|
|
+ const payload = await buildFlowisePayload(event, customAttributes);
|
|
|
+ if (typeof customAttributes.flowise_session_id !== 'string') {
|
|
|
+ try {
|
|
|
+ await chatwoot.setCustomAttributes(event.conversationId, {
|
|
|
+ ...customAttributes,
|
|
|
+ flowise_session_id: String(payload.overrideConfig.sessionId),
|
|
|
+ });
|
|
|
+ } catch {
|
|
|
+ logger.warn('Could not persist flowise_session_id in Chatwoot (local state still exists)', { conversationId: event.conversationId });
|
|
|
+ }
|
|
|
+ }
|
|
|
+ let response;
|
|
|
+ try {
|
|
|
+ response = await flowise.predict(payload);
|
|
|
+ } catch (err) {
|
|
|
+ const code = classifyAiError(err);
|
|
|
+ await handleAiFailure(event, chatwoot, code, err);
|
|
|
+ return { action: 'ai_error', code };
|
|
|
+ }
|
|
|
|
|
|
await audit({
|
|
|
conversationId: event.conversationId,
|
|
|
@@ -187,7 +264,10 @@ function isAssignedToRequiredAgentBot(
|
|
|
* Flowise payload. `[CONTACT_INFO]` is a contract with the deployed custom
|
|
|
* tools: `get_order_data` parses `contact_email:` out of `$flow.input`.
|
|
|
*/
|
|
|
-export function buildFlowisePayload(event: SupportMessageEvent): FlowisePayload {
|
|
|
+export async function buildFlowisePayload(
|
|
|
+ event: SupportMessageEvent,
|
|
|
+ customAttributes: Record<string, unknown> = event.customAttributes,
|
|
|
+): Promise<FlowisePayload> {
|
|
|
const cfg = config();
|
|
|
|
|
|
const contactLines: string[] = ['[CONTACT_INFO]'];
|
|
|
@@ -199,11 +279,12 @@ export function buildFlowisePayload(event: SupportMessageEvent): FlowisePayload
|
|
|
contactLines.push('[/CONTACT_INFO]');
|
|
|
|
|
|
const question = `${contactLines.join('\n')}\n\n---\n${event.content}`;
|
|
|
+ const { sessionId, created } = await getFlowiseSessionId(event.conversationId, customAttributes);
|
|
|
|
|
|
return {
|
|
|
question,
|
|
|
overrideConfig: {
|
|
|
- sessionId: `chatwoot:${event.conversationId}`,
|
|
|
+ sessionId,
|
|
|
conversationId: event.conversationId,
|
|
|
inboxId: event.inboxId,
|
|
|
messageId: event.messageId,
|
|
|
@@ -219,6 +300,64 @@ export function buildFlowisePayload(event: SupportMessageEvent): FlowisePayload
|
|
|
labels: event.labels,
|
|
|
subject: event.subject,
|
|
|
attachments: event.attachmentCount,
|
|
|
+ flowiseSessionPersisted: !created,
|
|
|
+ formMail: event.formMail
|
|
|
+ ? { technicalSender: event.formMail.technicalSender, customerEmail: event.formMail.customerEmail }
|
|
|
+ : null,
|
|
|
},
|
|
|
};
|
|
|
}
|
|
|
+
|
|
|
+function classifyAiError(err: unknown): string {
|
|
|
+ if (err instanceof RelayError) {
|
|
|
+ if (err.code === 'UPSTREAM_TIMEOUT') return 'FLOWISE_TIMEOUT';
|
|
|
+ if (/quota|insufficient/i.test(err.message)) return 'OPENAI_INSUFFICIENT_QUOTA';
|
|
|
+ if (/rate.?limit|429/i.test(err.message)) return 'OPENAI_RATE_LIMIT';
|
|
|
+ if (/invalid json|doctype|html/i.test(err.message)) return 'FLOWISE_INVALID_JSON';
|
|
|
+ if (err.code.includes('FLOWISE')) return err.code;
|
|
|
+ return err.code;
|
|
|
+ }
|
|
|
+ const msg = err instanceof Error ? err.message : String(err);
|
|
|
+ if (/timeout|abort/i.test(msg)) return 'FLOWISE_TIMEOUT';
|
|
|
+ if (/insufficient_quota|quota|billing|balance/i.test(msg)) return 'OPENAI_INSUFFICIENT_QUOTA';
|
|
|
+ if (/rate.?limit|429/i.test(msg)) return 'OPENAI_RATE_LIMIT';
|
|
|
+ return 'FLOWISE_ERROR';
|
|
|
+}
|
|
|
+
|
|
|
+async function handleAiFailure(
|
|
|
+ event: SupportMessageEvent,
|
|
|
+ chatwoot: ChatwootClient,
|
|
|
+ code: string,
|
|
|
+ err: unknown,
|
|
|
+): Promise<void> {
|
|
|
+ const terminal = /INSUFFICIENT_QUOTA|AUTH|INVALID|401|403/.test(code);
|
|
|
+ const now = new Date().toISOString();
|
|
|
+ const safeMessage = err instanceof Error ? err.message.slice(0, 300) : String(err).slice(0, 300);
|
|
|
+ await setMessageStatus(event.source, event.messageId, 'failed', code);
|
|
|
+ await audit({
|
|
|
+ conversationId: event.conversationId,
|
|
|
+ messageId: event.messageId,
|
|
|
+ eventType: 'ai_error',
|
|
|
+ summary: `AI/Flowise failure: ${code}`,
|
|
|
+ meta: { code, terminal, error: safeMessage },
|
|
|
+ });
|
|
|
+ try {
|
|
|
+ await chatwoot.addLabel(event.conversationId, 'ai-error');
|
|
|
+ if (terminal) await chatwoot.addLabel(event.conversationId, 'ticket');
|
|
|
+ await chatwoot.setCustomAttributes(event.conversationId, {
|
|
|
+ ai_status: 'error',
|
|
|
+ ai_error_code: code,
|
|
|
+ ai_error_at: now,
|
|
|
+ ai_last_failed_message_id: event.messageId,
|
|
|
+ ...(terminal ? { handoff: true, handoff_reason: code } : {}),
|
|
|
+ });
|
|
|
+ await chatwoot.sendPrivateNote(
|
|
|
+ event.conversationId,
|
|
|
+ `AI nie odpowiedziała na wiadomość ${event.messageId}. Kod: ${code}. Sprawdź klienta ręcznie; relay nie wysłał duplikatu odpowiedzi do klienta.`,
|
|
|
+ );
|
|
|
+ await chatwoot.unassignConversation(event.conversationId);
|
|
|
+ await chatwoot.setConversationStatus(event.conversationId, 'open');
|
|
|
+ } catch {
|
|
|
+ logger.warn('Could not persist AI failure side effects in Chatwoot', { conversationId: event.conversationId, code });
|
|
|
+ }
|
|
|
+}
|