From 61045aabc86ca059b8a9f9a13b66b2cce9bf9aee Mon Sep 17 00:00:00 2001 From: MythEclipse Date: Wed, 27 May 2026 23:32:38 +0700 Subject: [PATCH] fix(moderation): patch 6 aiAnalyzer audit vulnerabilities MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit #1+#5 - Individual fallback circuit breaker - Add individualConsecutiveErrors + individualCooldownUntil (30s) - On success: reset counter; on failure: increment + trip at AI_ANALYSIS_INDIVIDUAL_CB_THRESHOLD (default 10) consecutive errors - enqueueIndividualFallbacks checks CB before admitting any work #1 - Individual fallback concurrency cap - enqueueIndividualFallbacks enforces AI_ANALYSIS_INDIVIDUAL_MAX_CONCURRENT (default 20); overflow stays as error/analysis_incomplete in DB and is recovered by the recovery worker on the next interval #3 - Unhandled rejection in async setTimeout - scheduleConversationAnalysis no longer uses async arrow in setTimeout; all async work is chained with .then()/.catch() explicitly #4 - Recovery worker ignores individualInFlight - Add individualInFlightByConversation Map - processIndividualFallback increments/decrements it in try/finally - startPendingAIAnalysisWorker skips conversations present in the map - Recovery worker also processes error/analysis_incomplete messages via two new messageStore queries: getConversationKeysWithIncompleteAnalysis and getIncompleteMessagesByConversation #6 - pickBatchWithinBudget never called - scheduleConversationAnalysis now calls pickBatchWithinBudget with AI_ANALYSIS_MAX_TARGET_TOKENS (default 4000) + 50-token per-msg overhead after fetching messages, before passing to processBatch #7 - AI_PROCESSING_OVERLAP_MS 30s shorter than max LLM retry window - Replace hardcoded 30 000 ms constant with configurable AI_ANALYSIS_PROCESSING_TIMEOUT_MS (default 120 000 ms) - LLM client: 30s timeout × 3 retries + backoff ≈ 90-100s; 120s is safe New config keys: AI_ANALYSIS_PROCESSING_TIMEOUT_MS (default: 120000) AI_ANALYSIS_INDIVIDUAL_MAX_CONCURRENT (default: 20) AI_ANALYSIS_INDIVIDUAL_CB_THRESHOLD (default: 10) AI_ANALYSIS_MAX_TARGET_TOKENS (default: 4000) New AnalysisQueueStatus fields: activeIndividualRequests, individualInFlightCount, individualCircuitBreakerActive --- src/config.ts | 29 +++ src/moderation/aiAnalyzer.ts | 342 ++++++++++++++++++++++++--------- src/moderation/messageStore.ts | 85 ++++++++ src/moderation/types.ts | 2 + 4 files changed, 371 insertions(+), 87 deletions(-) diff --git a/src/config.ts b/src/config.ts index 1f95b70..7fcf0b5 100644 --- a/src/config.ts +++ b/src/config.ts @@ -80,11 +80,40 @@ const configSchema = z AI_ANALYSIS_ERROR_COOLDOWN_MS: z.coerce.number().positive().default(30000), AI_ANALYSIS_MAX_BATCH_SIZE: z.coerce.number().int().positive().default(25), AI_ANALYSIS_MAX_CONTEXT_TOKENS: z.coerce.number().positive().default(8000), + /** Token budget for target messages specifically (separate from context window). */ + AI_ANALYSIS_MAX_TARGET_TOKENS: z.coerce.number().positive().default(4000), AI_ANALYSIS_CONTEXT_MESSAGE_LIMIT: z.coerce .number() .int() .positive() .default(20), + /** + * How long a conversation is considered locked while being processed. + * Must exceed (LLM timeout × max retries) + network overhead. + * LLM client timeout=30s, retries=3 → minimum safe value ≈ 100s. + */ + AI_ANALYSIS_PROCESSING_TIMEOUT_MS: z.coerce + .number() + .positive() + .default(120000), + /** + * Maximum number of concurrent individual-fallback LLM calls. + * Prevents OOM/connection exhaustion when many messages miss a batch. + */ + AI_ANALYSIS_INDIVIDUAL_MAX_CONCURRENT: z.coerce + .number() + .int() + .positive() + .default(20), + /** + * How many consecutive individual-fallback errors trigger the individual + * circuit breaker (separate from the batch circuit breaker). + */ + AI_ANALYSIS_INDIVIDUAL_CB_THRESHOLD: z.coerce + .number() + .int() + .positive() + .default(10), DATABASE_TYPE: z.enum(["sqlite", "postgres"]).default("sqlite"), DATABASE_URL: z.string().optional(), POSTGRES_HOST: z.string().default("localhost"), diff --git a/src/moderation/aiAnalyzer.ts b/src/moderation/aiAnalyzer.ts index 974cfae..b37cdc0 100644 --- a/src/moderation/aiAnalyzer.ts +++ b/src/moderation/aiAnalyzer.ts @@ -13,6 +13,8 @@ import { runModerationAnalysis } from "./llmModerationClient.js"; import { getAttachmentsForMessages, getConversationContextBefore, + getConversationKeysWithIncompleteAnalysis, + getIncompleteMessagesByConversation, getMessageById, getPendingConversationKeys, getPendingMessagesByConversation, @@ -34,19 +36,21 @@ function getModerationBroadcaster(): ModerationBroadcaster | undefined { return (globalThis as ModerationGlobal).moderationBroadcaster; } -// Debounce state per conversation key -const conversationDebounceTimers = new Map(); -// Track conversations currently being processed -const conversationProcessing = new Map(); -// Track conversations in error cooldown (failed recently) -const conversationErrorCooldown = new Map(); +// --------------------------------------------------------------------------- +// Batch pipeline state +// --------------------------------------------------------------------------- -const AI_PROCESSING_OVERLAP_MS = 30000; +/** Debounce timer handle per conversation key. */ +const conversationDebounceTimers = new Map(); +/** Timestamp of when processing started per conversation key. */ +const conversationProcessing = new Map(); +/** Cooldown expiry timestamp per conversation key after an error. */ +const conversationErrorCooldown = new Map(); let activeRequests = 0; let lastError: string | null = null; -// Global circuit breaker state +// Batch circuit breaker let consecutiveErrors = 0; const MAX_CONSECUTIVE_ERRORS = 5; let globalCooldownUntil = 0; @@ -54,18 +58,38 @@ let globalCooldownUntil = 0; // --------------------------------------------------------------------------- // Individual fallback queue — runs PARALLEL to the batch pipeline. // -// When a batch LLM call returns but some message IDs are absent from the -// response (analysis_incomplete), those IDs are enqueued here. Each message -// is processed independently and concurrently: there is no serialisation -// per-conversation, and a dedup Set prevents the same ID being in-flight twice. +// Design guarantees: +// • Concurrency is capped at config.AI_ANALYSIS_INDIVIDUAL_MAX_CONCURRENT. +// • A flat Set de-duplicates so the same message can't be +// in-flight twice (Discord snowflakes are globally unique, but be safe). +// • A Map lets the recovery worker skip conversations +// that already have individual work in progress (#4 fix). +// • A separate circuit breaker prevents a cascade of individual failures +// from hammering a down/rate-limited LLM endpoint (#1+#5 fix). // --------------------------------------------------------------------------- -/** IDs currently being processed one-by-one (in-flight or waiting to start). */ +/** IDs currently being processed one-by-one. */ const individualInFlight = new Set(); -/** Counter for observability (mirrors activeRequests but for individual path). */ +/** + * Per-conversation count of in-flight individual messages. + * Used by the recovery worker to avoid re-scheduling a conversation that + * already has individual fallback work running for it. + */ +const individualInFlightByConversation = new Map(); + +/** Counter for observability. */ let activeIndividualRequests = 0; +// Individual fallback circuit breaker (independent of batch CB) +let individualConsecutiveErrors = 0; +let individualCooldownUntil = 0; +const INDIVIDUAL_COOLDOWN_MS = 30000; + +// --------------------------------------------------------------------------- +// Piscina worker pool (batch path only) +// --------------------------------------------------------------------------- + function getAnalysisWorkerUrl(): URL { const candidates = [ new URL("./aiAnalysisWorker.js", import.meta.url), @@ -94,15 +118,20 @@ interface AnalysisWorkerResponse { error?: string; } +// --------------------------------------------------------------------------- +// Exported helpers +// --------------------------------------------------------------------------- + /** - * Gets the conversation key for a message (thread_id or channel_id) + * Gets the conversation key for a message (thread_id or channel_id). */ export function getConversationKey(message: MessageRecord): string { return message.thread_id || message.channel_id; } /** - * Picks a batch of messages within token budget + * Picks a batch of messages within a token budget. + * `tokensPerMessage` accounts for JSON structure overhead around each entry. */ export function pickBatchWithinBudget( messages: MessageRecord[], @@ -125,26 +154,44 @@ export function pickBatchWithinBudget( return batch; } +// --------------------------------------------------------------------------- +// Conversation lock helpers +// --------------------------------------------------------------------------- + function isConversationProcessingLocked(conversationKey: string): boolean { const startedAt = conversationProcessing.get(conversationKey); + // FIX #7: use configurable timeout that exceeds (LLM timeout × max retries). + // Old hardcoded value was 30 000 ms — shorter than a single LLM call under retries. return Boolean( - startedAt && Date.now() - startedAt < AI_PROCESSING_OVERLAP_MS, + startedAt && + Date.now() - startedAt < config.AI_ANALYSIS_PROCESSING_TIMEOUT_MS, ); } +// --------------------------------------------------------------------------- +// Individual fallback pipeline +// --------------------------------------------------------------------------- + /** - * Processes a batch of messages for a conversation - */ -/** - * Processes a single message through the LLM moderation pipeline directly - * (no worker pool — avoids IPC overhead for a single-item call). Called from - * the individual fallback queue; never from the batch path. + * Processes a single message directly in the main process (no IPC/worker + * pool overhead). Never called from the batch path. + * + * FIX #1+#5: Increments the individual circuit breaker on failure so a + * sustained outage stops hammering the LLM endpoint. */ async function processIndividualFallback( message: MessageRecord, ): Promise { const { id: messageId } = message; + const conversationKey = getConversationKey(message); + activeIndividualRequests++; + // Increment per-conversation counter so the recovery worker can see it. + individualInFlightByConversation.set( + conversationKey, + (individualInFlightByConversation.get(conversationKey) ?? 0) + 1, + ); + try { const contextBefore = await getConversationContextBefore({ channelId: message.channel_id, @@ -198,11 +245,29 @@ async function processIndividualFallback( getModerationBroadcaster()?.messageAnalyzed(row); } + // Reset individual CB on success. + individualConsecutiveErrors = 0; + logger.info( { messageId, status: analysisResult.results[0]?.status }, "Individual fallback analysis complete", ); } catch (error) { + // FIX #5: individual failures now feed their own circuit breaker. + individualConsecutiveErrors++; + if ( + individualConsecutiveErrors >= config.AI_ANALYSIS_INDIVIDUAL_CB_THRESHOLD + ) { + individualCooldownUntil = Date.now() + INDIVIDUAL_COOLDOWN_MS; + logger.warn( + { + threshold: config.AI_ANALYSIS_INDIVIDUAL_CB_THRESHOLD, + cooldownUntil: new Date(individualCooldownUntil).toISOString(), + }, + "Individual fallback circuit breaker triggered", + ); + } + lastError = error instanceof Error ? error.message : String(error); logger.error( { @@ -215,48 +280,99 @@ async function processIndividualFallback( } finally { activeIndividualRequests--; individualInFlight.delete(messageId); + + // Decrement per-conversation counter; remove key when it hits zero. + const prev = individualInFlightByConversation.get(conversationKey) ?? 1; + if (prev <= 1) { + individualInFlightByConversation.delete(conversationKey); + } else { + individualInFlightByConversation.set(conversationKey, prev - 1); + } } } /** - * Fans out a list of message records to the individual fallback queue. - * Each message starts processing concurrently (fire-and-forget per message). - * De-duplicated by message ID so no double-processing even if called repeatedly. + * Fans out message records to the individual fallback queue. + * + * FIX #1: Checks concurrency cap before admitting new work. + * FIX #5: Checks individual circuit breaker before admitting new work. + * Messages that cannot be admitted remain as `error/analysis_incomplete` in + * the DB and will be picked up by the recovery worker on the next interval. */ function enqueueIndividualFallbacks(messages: MessageRecord[]): void { + // FIX #5: Honour the individual circuit breaker. + if (Date.now() < individualCooldownUntil) { + logger.warn( + { + until: new Date(individualCooldownUntil).toISOString(), + skipped: messages.length, + }, + "Individual fallback circuit breaker active — messages will be recovered later", + ); + return; + } + const newMessages = messages.filter((m) => !individualInFlight.has(m.id)); if (newMessages.length === 0) return; + // FIX #1: Enforce concurrency cap. + const availableSlots = + config.AI_ANALYSIS_INDIVIDUAL_MAX_CONCURRENT - individualInFlight.size; + if (availableSlots <= 0) { + logger.warn( + { + cap: config.AI_ANALYSIS_INDIVIDUAL_MAX_CONCURRENT, + inFlight: individualInFlight.size, + skipped: newMessages.length, + }, + "Individual fallback concurrency cap reached — messages will be recovered by recovery worker", + ); + return; + } + + const toProcess = newMessages.slice(0, availableSlots); + const skipped = newMessages.length - toProcess.length; + logger.info( { - count: newMessages.length, - messageIds: newMessages.map((m) => m.id), + count: toProcess.length, + skipped, + messageIds: toProcess.map((m) => m.id), }, "Enqueueing individual fallback analysis for batch-incomplete messages", ); - for (const msg of newMessages) { + for (const msg of toProcess) { individualInFlight.add(msg.id); - // Fire-and-forget: each message runs concurrently, errors are handled inside. + // Fire-and-forget: processIndividualFallback handles all errors internally. processIndividualFallback(msg).catch((err) => { - // Belt-and-suspenders: processIndividualFallback catches internally, - // but guard against any uncaught rejection bubbling here. + // Belt-and-suspenders guard — should never reach here. logger.error( { messageId: msg.id, error: String(err) }, - "Unexpected error in individual fallback promise", + "Unexpected uncaught error escaping processIndividualFallback", ); individualInFlight.delete(msg.id); + const ck = getConversationKey(msg); + const prev = individualInFlightByConversation.get(ck) ?? 1; + if (prev <= 1) { + individualInFlightByConversation.delete(ck); + } else { + individualInFlightByConversation.set(ck, prev - 1); + } }); } } +// --------------------------------------------------------------------------- +// Batch pipeline +// --------------------------------------------------------------------------- + async function processBatch( conversationKey: string, messages: MessageRecord[], ): Promise { if (messages.length === 0) return; if (Date.now() < globalCooldownUntil) { - // Should not normally hit here due to checks in scheduleConversationAnalysis, but just in case return; } @@ -348,7 +464,7 @@ async function processBatch( enqueueIndividualFallbacks(incompleteMessages); } - consecutiveErrors = 0; // Reset circuit breaker + consecutiveErrors = 0; // Reset batch circuit breaker conversationErrorCooldown.delete(conversationKey); shouldScheduleNext = true; } catch (error) { @@ -396,76 +512,96 @@ async function processBatch( } } +// --------------------------------------------------------------------------- +// Scheduling +// --------------------------------------------------------------------------- + /** - * Debounced analysis trigger for a conversation + * Schedules a debounced analysis run for a conversation. + * + * FIX #3: The async work inside setTimeout is now wrapped in an explicit + * .catch() so DB errors don't produce unhandled promise rejections. + * FIX #6: Calls pickBatchWithinBudget after fetching messages so token budget + * is respected before handing the batch to the LLM. */ function scheduleConversationAnalysis(conversationKey: string): void { - // Skip if already processing if (isConversationProcessingLocked(conversationKey)) { return; } - // Check cooldowns const convoCooldown = conversationErrorCooldown.get(conversationKey) || 0; const activeCooldown = Math.max(convoCooldown, globalCooldownUntil); if (activeCooldown && Date.now() < activeCooldown) { - // Instead of dropping, re-schedule for after cooldown if not already scheduled if (!conversationDebounceTimers.has(conversationKey)) { const remaining = activeCooldown - Date.now(); const timer = setTimeout(() => { conversationDebounceTimers.delete(conversationKey); scheduleConversationAnalysis(conversationKey); - }, remaining + 500); // 500ms buffer after cooldown + }, remaining + 500); conversationDebounceTimers.set(conversationKey, timer); } return; } - // Clear existing timer const existingTimer = conversationDebounceTimers.get(conversationKey); if (existingTimer) { clearTimeout(existingTimer); } - // Always use shorter debounce for immediate processing (no concurrency limit) - const debounceTime = config.AI_ANALYSIS_DEBOUNCE_MS; - - // Set new debounced timer - const timer = setTimeout(async () => { + const timer = setTimeout(() => { conversationDebounceTimers.delete(conversationKey); - // Get pending messages for this conversation - const messages = await getPendingMessagesByConversation( + // FIX #3: explicit .catch() — no async arrow function to avoid unhandled rejection. + getPendingMessagesByConversation( conversationKey, config.AI_ANALYSIS_MAX_BATCH_SIZE, - ); + ) + .then((messages) => { + if (messages.length === 0) return; - if (messages.length > 0) { - await processBatch(conversationKey, messages); - } - }, debounceTime); + // FIX #6: trim to token budget before sending to LLM. + // 50 tokens overhead accounts for JSON structure + id/username fields. + const trimmed = pickBatchWithinBudget( + messages, + config.AI_ANALYSIS_MAX_TARGET_TOKENS, + 50, + ); + if (trimmed.length === 0) return; + + return processBatch(conversationKey, trimmed); + }) + .catch((err) => { + logger.error( + { + conversationKey, + error: err instanceof Error ? err.message : String(err), + }, + "Failed to fetch or dispatch pending messages for scheduled analysis", + ); + }); + }, config.AI_ANALYSIS_DEBOUNCE_MS); conversationDebounceTimers.set(conversationKey, timer); } +// --------------------------------------------------------------------------- +// Public API +// --------------------------------------------------------------------------- + /** - * Queues a message for analysis (debounced by conversation) + * Queues a message for analysis (debounced by conversation). */ export async function queueMessageAnalysis(messageId: string): Promise { if (!config.AI_ANALYSIS_ENABLED) return; try { - // Look up the message to get its conversation key const message = await getMessageById(messageId); if (!message) { logger.warn({ messageId }, "Message not found for analysis queue"); return; } - - // Schedule its conversation for analysis - const conversationKey = getConversationKey(message); - queueConversationAnalysis(conversationKey); + queueConversationAnalysis(getConversationKey(message)); } catch (error) { logger.error( { @@ -478,17 +614,15 @@ export async function queueMessageAnalysis(messageId: string): Promise { } /** - * Queues a conversation for analysis (debounced) + * Queues a conversation for analysis (debounced). */ export function queueConversationAnalysis(conversationKey: string): void { if (!config.AI_ANALYSIS_ENABLED) return; - - // Schedule debounced analysis scheduleConversationAnalysis(conversationKey); } /** - * Gets current analysis queue status + * Returns current status of both the batch and individual fallback queues. */ export function getAnalysisQueueStatus(): AnalysisQueueStatus { return { @@ -496,42 +630,76 @@ export function getAnalysisQueueStatus(): AnalysisQueueStatus { activeRequests, activeIndividualRequests, individualInFlightCount: individualInFlight.size, + individualCircuitBreakerActive: Date.now() < individualCooldownUntil, lastError, }; } /** - * Starts the pending AI analysis recovery worker + * Starts the periodic recovery worker. + * + * FIX #4: Now also recovers messages stuck in `error/analysis_incomplete` + * state (not just `pending`), and skips conversations that already have + * individual fallback work in progress to avoid DB last-write-wins races. */ export function startPendingAIAnalysisWorker(): void { if (!config.AI_ANALYSIS_ENABLED) return; - setInterval(async () => { - try { - // Get pending conversation keys - const conversationKeys = await getPendingConversationKeys(100); - - for (const key of conversationKeys) { - // Skip if already scheduled - if (conversationDebounceTimers.has(key)) { - continue; + setInterval(() => { + // FIX #3 pattern: no async arrow — chain promises explicitly. + Promise.all([ + getPendingConversationKeys(100), + getConversationKeysWithIncompleteAnalysis(50), + ]) + .then(([pendingKeys, incompleteKeys]) => { + // --- Batch recovery for `pending` messages --- + for (const key of pendingKeys) { + if (conversationDebounceTimers.has(key)) continue; + if (isConversationProcessingLocked(key)) continue; + // FIX #4: skip if individual fallback already running for this conversation. + if (individualInFlightByConversation.has(key)) continue; + const cooldownUntil = conversationErrorCooldown.get(key); + if (cooldownUntil && Date.now() < cooldownUntil) continue; + scheduleConversationAnalysis(key); } - // Skip if currently processing - if (isConversationProcessingLocked(key)) { - continue; - } + // --- Individual recovery for `error/analysis_incomplete` messages --- + // Circuit breaker check: no point iterating if individual CB is active. + if (Date.now() >= individualCooldownUntil) { + const promises: Promise[] = []; + for (const key of incompleteKeys) { + // Skip if individual work is already running for this conversation. + if (individualInFlightByConversation.has(key)) continue; + // Skip if batch processing is running (it will fan-out if it finds more incomplete). + if (isConversationProcessingLocked(key)) continue; - // Skip if in error cooldown - const cooldownUntil = conversationErrorCooldown.get(key); - if (cooldownUntil && Date.now() < cooldownUntil) { - continue; + promises.push( + getIncompleteMessagesByConversation( + key, + config.AI_ANALYSIS_INDIVIDUAL_MAX_CONCURRENT, + ) + .then((msgs) => { + if (msgs.length > 0) { + enqueueIndividualFallbacks(msgs); + } + }) + .catch((err) => { + logger.error( + { key, error: String(err) }, + "Failed to fetch incomplete messages for recovery", + ); + }), + ); + } + // Errors are handled per-key; return the combined promise for observability. + return Promise.all(promises); } - - scheduleConversationAnalysis(key); - } - } catch (error) { - logger.error({ error }, "Pending AI analysis recovery worker failed"); - } + }) + .catch((err) => { + logger.error( + { error: err instanceof Error ? err.message : String(err) }, + "Pending AI analysis recovery worker failed", + ); + }); }, config.AI_ANALYSIS_RECOVERY_INTERVAL_MS); } diff --git a/src/moderation/messageStore.ts b/src/moderation/messageStore.ts index 6776b78..206ebfc 100644 --- a/src/moderation/messageStore.ts +++ b/src/moderation/messageStore.ts @@ -715,3 +715,88 @@ export async function searchMessages(input: { throw error; } } + +/** + * Returns distinct conversation keys (thread_id or channel_id) that have at + * least one message stuck in `error` status with the `analysis_incomplete` + * flag set. Used by the recovery worker to re-feed those messages through + * the individual-fallback queue. + */ +export async function getConversationKeysWithIncompleteAnalysis( + limit: number = 50, +): Promise { + try { + const database = db(); + const rows = await database + .selectDistinct>({ + thread_id: messagesTable.thread_id, + channel_id: messagesTable.channel_id, + }) + .from(messagesTable) + .where( + and( + eq(messagesTable.ai_status, "error"), + sql`${messagesTable.ai_moderation_flags} LIKE ${"%analysis_incomplete%"}`, + isNull(messagesTable.deleted_at), + ), + ) + .limit(limit); + + const keys: string[] = []; + for (const row of rows) { + const key = row.thread_id || row.channel_id; + if (key && !keys.includes(key)) { + keys.push(key); + } + } + return keys; + } catch (error) { + logger.error( + { error: error instanceof Error ? error.message : String(error) }, + "Failed to get conversation keys with incomplete analysis", + ); + throw error; + } +} + +/** + * Returns MessageRecords for a given conversation key whose AI analysis is + * stuck in `error` + `analysis_incomplete`. Used to feed those records + * directly into the individual-fallback queue without touching their status + * (the individual pipeline will overwrite status on success). + */ +export async function getIncompleteMessagesByConversation( + conversationKey: string, + limit: number = 20, +): Promise { + try { + const database = db(); + const rows = await database + .select() + .from(messagesTable) + .where( + and( + or( + eq(messagesTable.thread_id, conversationKey), + eq(messagesTable.channel_id, conversationKey), + ), + eq(messagesTable.ai_status, "error"), + sql`${messagesTable.ai_moderation_flags} LIKE ${"%analysis_incomplete%"}`, + isNull(messagesTable.deleted_at), + ), + ) + .orderBy(asc(messagesTable.created_at)) + .limit(limit); + + return rows as MessageRecord[]; + } catch (error) { + logger.error( + { + conversationKey, + error: error instanceof Error ? error.message : String(error), + }, + "Failed to get incomplete messages by conversation", + ); + throw error; + } +} diff --git a/src/moderation/types.ts b/src/moderation/types.ts index b7c0580..b8cdd54 100644 --- a/src/moderation/types.ts +++ b/src/moderation/types.ts @@ -141,5 +141,7 @@ export interface AnalysisQueueStatus { activeIndividualRequests: number; /** Number of message IDs sitting in the dedup set (in-flight or about to start). */ individualInFlightCount: number; + /** True when the individual-fallback circuit breaker is tripped. */ + individualCircuitBreakerActive: boolean; lastError: string | null; }