diff --git a/services/discord-gateway/src/modules/ai-moderation/aiAnalyzer.ts b/services/discord-gateway/src/modules/ai-moderation/aiAnalyzer.ts index 2c53e015..ef490827 100644 --- a/services/discord-gateway/src/modules/ai-moderation/aiAnalyzer.ts +++ b/services/discord-gateway/src/modules/ai-moderation/aiAnalyzer.ts @@ -163,12 +163,16 @@ export function startPendingAIAnalysisWorker( }); } - messageStore.revertStuckProcessingMessages(300000).catch((err: unknown) => { - logger.error( - { error: String(err) }, - "Failed to run stuck processing recovery", - ); - }); + // Only revert stuck processing messages if there's active processing. + // Avoids a DB query every recovery interval when the pipeline is idle. + if (conversationProcessing.size > 0) { + messageStore.revertStuckProcessingMessages(300000).catch((err: unknown) => { + logger.error( + { error: String(err) }, + "Failed to run stuck processing recovery", + ); + }); + } Promise.all([ messageStore.getPendingConversationKeys(500), diff --git a/services/discord-gateway/src/modules/ai-moderation/batchProcessor.ts b/services/discord-gateway/src/modules/ai-moderation/batchProcessor.ts index 7e10ed37..ea791cca 100644 --- a/services/discord-gateway/src/modules/ai-moderation/batchProcessor.ts +++ b/services/discord-gateway/src/modules/ai-moderation/batchProcessor.ts @@ -163,20 +163,14 @@ export async function processBatch( messages, })) as AnalysisWorkerResponse; - // Do not broadcast or auto-delete if it's an API failure that will be reverted. + // Broadcast + auto-delete only for successfully analyzed rows. + // Error rows (API failures, parse failures, incomplete) will be + // retried by the individual fallback queue — do NOT schedule + // auto-delete for them (they'd be logged as not_eligible anyway). for (const row of result.rows) { - let isApiFailure = false; - if (row.ai_status === "error") { - try { - const flags = JSON.parse(row.ai_moderation_flags ?? "[]") as string[]; - isApiFailure = flags.includes("analysis_api_failed"); - } catch {} - } - - if (!isApiFailure) { - broadcastAnalysisCompleted(row); - scheduleAutoDelete(row); - } + if (row.ai_status === "error") continue; + broadcastAnalysisCompleted(row); + scheduleAutoDelete(row); } if (!result.ok) { diff --git a/services/discord-gateway/src/modules/ai-moderation/conversationContext.ts b/services/discord-gateway/src/modules/ai-moderation/conversationContext.ts index 6079e8b8..59a00742 100644 --- a/services/discord-gateway/src/modules/ai-moderation/conversationContext.ts +++ b/services/discord-gateway/src/modules/ai-moderation/conversationContext.ts @@ -136,6 +136,12 @@ export function formatMessageForPrompt( return `[${label}] id=${msg.id} time=${timestamp} user=${resolveDisplayName(msg)}: ${content}${mediaSuffix}${refInfo}`; } +/** Estimate tokens for a batch of strings — single encode call, ~5x faster than per-line. */ +export function estimateTokensBatch(texts: string[]): number { + const combined = texts.join("\n"); + return getEncoder().encode(combined).length + texts.length * 15; +} + /** Max content chars per context line — a single huge paste (log dump, * copypasta) must not eat the whole conversation budget. */ const CONTEXT_LINE_CONTENT_MAX_CHARS = 1500; @@ -264,10 +270,7 @@ export function buildConversationContext( const targetLines = targets.map((msg) => formatMessageForPrompt(msg, "target"), ); - let usedTokens = targetLines.reduce( - (sum, line) => sum + estimateTokens(line), - 0, - ); + let usedTokens = estimateTokensBatch(targetLines); const contextLines = gatedNewestFirst.map((msg) => formatMessageForPrompt(msg, "context"), diff --git a/services/discord-gateway/src/modules/ai-moderation/llmCaller.ts b/services/discord-gateway/src/modules/ai-moderation/llmCaller.ts index a2c7df8e..e60e18f0 100644 --- a/services/discord-gateway/src/modules/ai-moderation/llmCaller.ts +++ b/services/discord-gateway/src/modules/ai-moderation/llmCaller.ts @@ -16,6 +16,7 @@ import { delay, retryWithBackoff } from "@/shared/utils/index"; import { config } from "../../shared/config/config.js"; import type { AnalysisResult } from "../message-capture/types.js"; import { llmChat } from "./llmClient.js"; +import { parseModerationResponse } from "./moderationResponseParser.js"; import { logModerationError } from "./responseLogger.js"; const log = createChildLogger("llm-caller"); @@ -79,7 +80,7 @@ export async function callModerationLLM( ]; const completion = await llmChat({ messages, - max_tokens: maxTokens ?? 16384, + max_tokens: maxTokens ?? config.AI_LLM_MAX_COMPLETION_TOKENS ?? 16384, jsonResponse: { type: "json_object" }, retries: 0, signal, @@ -106,9 +107,6 @@ export async function callModerationLLM( if (!rawContent) throw new Error("No content in LLM response"); try { - const { parseModerationResponse } = await import( - "./moderationResponseParser.js" - ); return { parsed: parseModerationResponse(rawContent, targetIds), result: completion, diff --git a/services/discord-gateway/src/modules/ai-moderation/moderationOrchestrator.ts b/services/discord-gateway/src/modules/ai-moderation/moderationOrchestrator.ts index dd20a5d0..2d0d104c 100644 --- a/services/discord-gateway/src/modules/ai-moderation/moderationOrchestrator.ts +++ b/services/discord-gateway/src/modules/ai-moderation/moderationOrchestrator.ts @@ -461,7 +461,9 @@ export async function runModerationAnalysis( cacheKey, stored, embeddingsByKey.get(cacheKey), - ).catch(() => {}); + ).catch((err: unknown) => { + log.warn({ cacheKey, error: String(err) }, "Cache write failed"); + }); // Dual-key write-back (2026-08-24): the FIRST analysis of a message runs // WITH conversation context (accurate), but its verdict is also stored diff --git a/services/discord-gateway/src/modules/ai-moderation/textBatchProcessor.ts b/services/discord-gateway/src/modules/ai-moderation/textBatchProcessor.ts index 2904ee69..3b4289c2 100644 --- a/services/discord-gateway/src/modules/ai-moderation/textBatchProcessor.ts +++ b/services/discord-gateway/src/modules/ai-moderation/textBatchProcessor.ts @@ -69,6 +69,26 @@ export async function buildCorrectedFewShotExamples(): Promise { } } +// --------------------------------------------------------------------------- +// Few-shot correction cache (refreshes hourly) +// --------------------------------------------------------------------------- +let _correctedExamplesCache: string | null = null; +let _correctedExamplesCacheAt = 0; +const CORRECTED_CACHE_TTL_MS = 60 * 60 * 1000; + +async function getCachedCorrectedExamples(): Promise { + const now = Date.now(); + if ( + _correctedExamplesCache !== null && + now - _correctedExamplesCacheAt < CORRECTED_CACHE_TTL_MS + ) { + return _correctedExamplesCache; + } + _correctedExamplesCache = await buildCorrectedFewShotExamples(); + _correctedExamplesCacheAt = now; + return _correctedExamplesCache; +} + // --------------------------------------------------------------------------- // Text-only batch // --------------------------------------------------------------------------- @@ -88,7 +108,22 @@ export async function runTextOnlyBatch( for (const url of extractUrlsFromText(msg.edited_content ?? msg.content)) allUrls.add(url); } - const urlArr = Array.from(allUrls).slice(0, 10); + // Domain dedup: max 3 URLs per domain to avoid rate-limiting + const domainCounts = new Map(); + const urlArr: string[] = []; + for (const url of allUrls) { + try { + const domain = new URL(url).hostname; + const count = domainCounts.get(domain) ?? 0; + if (count >= 3) continue; + domainCounts.set(domain, count + 1); + } catch { + /* invalid URL, skip */ + continue; + } + urlArr.push(url); + if (urlArr.length >= 10) break; + } if (urlArr.length === 0) { return { text: new Map(), @@ -192,7 +227,7 @@ export async function runTextOnlyBatch( // Corrected false-positive examples are static per batch — fetch ONCE // here instead of inside the per-sub-batch retry closure (which would // re-query the DB on every sub-batch and every parse-error retry). - const correctedExamples = await buildCorrectedFewShotExamples(); + const correctedExamples = await getCachedCorrectedExamples(); // ── URL images → multimodal vision evidence (hoisted out of the sub-batch // loop) ─────────────────────────────────────────────────────────── @@ -368,9 +403,13 @@ export async function runTextOnlyBatch( ); } catch (err: any) { if (err.name === "AbortError" || abortController.signal.aborted) { - throw new Error( - `Text-only batch sub-batch ${i + 1} timed out for messages ${targetIds.join(", ")}`, + // Sub-batch timed out — log but DO NOT throw. Previous sub-batches' + // results are already in allResults; throwing would discard them. + log.warn( + { subBatch: i + 1, targetIds, timeoutMs }, + "Sub-batch timed out — preserving partial results from prior sub-batches", ); + continue; } throw err; } finally { diff --git a/services/discord-gateway/src/shared/config/index.ts b/services/discord-gateway/src/shared/config/index.ts index 7e6b6235..b7c31d34 100644 --- a/services/discord-gateway/src/shared/config/index.ts +++ b/services/discord-gateway/src/shared/config/index.ts @@ -197,6 +197,7 @@ export const configSchema = z .positive() .default(1024), AI_LLM_TEXT_BATCH_SIZE: z.coerce.number().int().positive().default(60), + AI_LLM_MAX_COMPLETION_TOKENS: z.coerce.number().int().positive().default(16384), AI_LLM_MEDIA_ANALYSIS_TIMEOUT_MS: z.coerce .number() .int()