diff --git a/.env.example b/.env.example index c18f620..d3d8cfd 100644 --- a/.env.example +++ b/.env.example @@ -92,3 +92,10 @@ AUTO_DELETE_ALLOWED_SEVERITIES=critical,high,medium # Safe default: run migrations on startup before the app accepts traffic. AUTO_MIGRATE_ON_STARTUP=true +# Cache Model Versioning +# Bump this version when the vision/LLM model prompt changes significantly. +# Old cache entries with mismatched versions are automatically ignored, forcing fresh analysis. +# Format: "v" or "v--" +# Example progression: v1 → v2-2026-06-02-terminal-fix → v3-2026-06-15-new-model +# CACHE_MODEL_VERSION=v2-2026-06-02 + diff --git a/services/discord-gateway/src/modules/ai-moderation/llmModerationClient.ts b/services/discord-gateway/src/modules/ai-moderation/llmModerationClient.ts index 217fe80..1955fc7 100644 --- a/services/discord-gateway/src/modules/ai-moderation/llmModerationClient.ts +++ b/services/discord-gateway/src/modules/ai-moderation/llmModerationClient.ts @@ -32,8 +32,17 @@ import { makeImageCacheKey, makeStickerCacheKey, upsertCachedMediaAnalysis, + VISION_MODEL_VERSION, } from "./textCacheStore.js"; import { extractUrlsFromText, fetchUrlSafely } from "./urlFetcher.js"; +import { + logVisionAnalysis, + logCacheEvent, + logModerationAnalysis, + logModerationError, + logVisionError, + logRetryAttempt, +} from "./responseLogger.js"; const SeveritySchema = z.enum(["none", "low", "medium", "high", "critical"]); const RecommendedActionSchema = z.enum([ @@ -786,6 +795,13 @@ async function callModerationLLM( `Robust Fallback (${label}): Failed to parse moderation response. Marking all targets as analysis errors.`, ); + // Log error with responseLogger + logModerationError(targetIds, config.AI_LLM_MODEL, parseError, { + phase: "parse_response", + label, + contentLength: state.lastInvalidContent.length, + }); + // Sanitized error messages — no internal details exposed (R10) const errorCode = `MOD_${Date.now().toString(36).slice(0, 6)}`; parsed = targetIds.map((id) => ({ @@ -896,6 +912,21 @@ async function runTextOnlyBatch( allResults.push(...batchResult.results); if (batchResult.raw) lastRaw = batchResult.raw; + + // Log batch results with comprehensive details + logModerationAnalysis( + targetIds, + config.AI_LLM_MODEL, + batchResult.results, + 0, // Duration will be tracked at higher level + batchResult.raw?.usage + ? { + prompt_tokens: batchResult.raw.usage.prompt_tokens, + completion_tokens: batchResult.raw.usage.completion_tokens, + total_tokens: batchResult.raw.usage.total_tokens, + } + : undefined, + ); } log.info( @@ -1012,6 +1043,7 @@ async function _runSingleMediaAnalysis( { attachmentId: att.id, cacheKey: attVisionKey }, "Vision cache HIT for attachment — skipped download", ); + logCacheEvent("hit", attVisionKey, "media", VISION_MODEL_VERSION); const sourceLabel = `[gambar di atas adalah attachment ${att.filename} dari pesan id=${att.message_id}]`; const analysisText = `[Media analysis for message ${att.message_id}] ${sourceLabel}: ${cachedVision}`; const existing = mediaAnalysisMap.get(targetId) ?? []; @@ -1020,6 +1052,8 @@ async function _runSingleMediaAnalysis( return; } + logCacheEvent("miss", attVisionKey, "media", VISION_MODEL_VERSION); + const controller = new AbortController(); const timeoutId = setTimeout(() => controller.abort(), 15000); diff --git a/services/discord-gateway/src/modules/ai-moderation/responseLogger.ts b/services/discord-gateway/src/modules/ai-moderation/responseLogger.ts new file mode 100644 index 0000000..b817794 --- /dev/null +++ b/services/discord-gateway/src/modules/ai-moderation/responseLogger.ts @@ -0,0 +1,334 @@ +/** + * Comprehensive response logging for vision and moderation analysis. + * + * Purpose: Capture ALL responses from: + * - Vision model (image analysis) + * - LLM moderation (text analysis) + * - Cache hits/misses + * - Errors and retries + * + * Enables full audit trail and debugging of moderation decisions. + */ + +import { createChildLogger } from "../../shared/logger/logger.js"; +import type { AnalysisResult } from "../message-capture/types.js"; + +const logger = createChildLogger("response-logger"); + +export interface VisionAnalysisResponse { + messageId: string; + cacheKey: string; + cached: boolean; + description: string; + duration_ms: number; + timestamp: number; +} + +export interface ModerationAnalysisResponse { + messageIds: string[]; + batchSize: number; + model: string; + tokenUsage?: { + prompt_tokens: number; + completion_tokens: number; + total_tokens: number; + }; + results: AnalysisResult[]; + duration_ms: number; + parseErrors: string[]; + timestamp: number; +} + +export interface CacheHitEvent { + type: "hit" | "miss"; + cacheKey: string; + source: "text" | "media" | "sticker"; + modelVersion: string; + timestamp: number; +} + +/** + * Log a vision model response with full details. + */ +export function logVisionAnalysis( + messageId: string, + cacheKey: string, + cached: boolean, + description: string, + duration_ms: number, +): void { + const response: VisionAnalysisResponse = { + messageId, + cacheKey, + cached, + description, + duration_ms, + timestamp: Date.now(), + }; + + logger.info( + { + ...response, + description_length: description.length, + description_preview: description.substring(0, 200), + }, + `Vision analysis complete [${cached ? "CACHED" : "FRESH"}] for message ${messageId}`, + ); +} + +/** + * Log a moderation LLM response with full batch details. + */ +export function logModerationAnalysis( + messageIds: string[], + model: string, + results: AnalysisResult[], + duration_ms: number, + tokenUsage?: { prompt_tokens: number; completion_tokens: number; total_tokens: number }, + parseErrors: string[] = [], +): void { + const response: ModerationAnalysisResponse = { + messageIds, + batchSize: messageIds.length, + model, + tokenUsage, + results: results.map((r) => ({ + messageId: r.messageId, + status: r.status, + flags: r.flags ?? [], + score: r.score, + severity: r.severity, + confidence: r.confidence, + recommendedAction: r.recommendedAction, + analysis: r.analysis?.substring(0, 200), // Truncate for logs + })) as AnalysisResult[], + duration_ms, + parseErrors, + timestamp: Date.now(), + }; + + logger.info( + { + batch_size: messageIds.length, + model, + token_usage: tokenUsage, + duration_ms, + parse_errors: parseErrors.length, + results_summary: { + clean: results.filter((r) => r.status === "clean").length, + warn: results.filter((r) => r.status === "warn").length, + flagged: results.filter((r) => r.status === "flagged").length, + error: results.filter((r) => r.status === "error").length, + }, + }, + `Moderation analysis complete for batch of ${messageIds.length} messages`, + ); + + // Log each result individually for detailed audit trail + results.forEach((result, idx) => { + const severity = result.severity ?? "none"; + const confidence = result.confidence ?? 0; + logger.debug( + { + index: idx, + message_id: result.messageId, + status: result.status, + flags: result.flags, + score: result.score, + severity, + confidence, + categories: result.categories, + recommended_action: result.recommendedAction, + analysis: result.analysis?.substring(0, 300), + evidence: result.evidence?.slice(0, 3), // First 3 evidence items + }, + `[${idx + 1}/${messageIds.length}] Moderation result for message ${result.messageId}: ${result.status} (severity: ${severity}, confidence: ${confidence})`, + ); + }); +} + +/** + * Log cache hit/miss event. + */ +export function logCacheEvent( + type: "hit" | "miss", + cacheKey: string, + source: "text" | "media" | "sticker", + modelVersion: string, +): void { + const event: CacheHitEvent = { + type, + cacheKey, + source, + modelVersion, + timestamp: Date.now(), + }; + + const level = type === "hit" ? "debug" : "debug"; + logger.log( + { level }, + { + cache_type: type.toUpperCase(), + source, + key_length: cacheKey.length, + key_preview: cacheKey.substring(0, 50), + model_version: modelVersion, + }, + `Cache ${type.toUpperCase()}: ${source} (version: ${modelVersion})`, + ); +} + +/** + * Log vision API error with context. + */ +export function logVisionError( + messageId: string, + error: Error | string, + context?: Record, +): void { + logger.error( + { + message_id: messageId, + error: error instanceof Error ? error.message : String(error), + stack: error instanceof Error ? error.stack : undefined, + context, + }, + `Vision analysis failed for message ${messageId}`, + ); +} + +/** + * Log LLM API error with request context. + */ +export function logModerationError( + messageIds: string[], + model: string, + error: Error | string, + context?: Record, +): void { + logger.error( + { + message_ids: messageIds, + batch_size: messageIds.length, + model, + error: error instanceof Error ? error.message : String(error), + stack: error instanceof Error ? error.stack : undefined, + context, + }, + `Moderation analysis failed for batch of ${messageIds.length} messages`, + ); +} + +/** + * Log retry event with attempt details. + */ +export function logRetryAttempt( + label: string, + attempt: number, + maxRetries: number, + error: Error | string, + nextDelayMs: number, +): void { + logger.warn( + { + label, + attempt, + max_retries: maxRetries, + error: error instanceof Error ? error.message : String(error), + next_delay_ms: nextDelayMs, + remaining_attempts: maxRetries - attempt + 1, + }, + `Retry attempt ${attempt}/${maxRetries} for ${label} (next retry in ${nextDelayMs}ms)`, + ); +} + +/** + * Log analysis completion summary (for batch end-of-processing). + */ +export function logAnalysisSummary( + conversationKey: string, + totalMessages: number, + successCount: number, + errorCount: number, + durationMs: number, + summary: Record, +): void { + logger.info( + { + conversation_key: conversationKey, + total_messages: totalMessages, + success_count: successCount, + error_count: errorCount, + duration_ms: durationMs, + per_message_avg_ms: Math.round(durationMs / totalMessages), + summary, + success_rate: ((successCount / totalMessages) * 100).toFixed(1) + "%", + }, + `Analysis batch complete: ${successCount}/${totalMessages} successful in ${durationMs}ms`, + ); +} + +/** + * Log false positive detection (cache version mismatch or incorrect analysis). + */ +export function logFalsePositiveDetected( + messageId: string, + currentAnalysis: AnalysisResult, + reason: string, + context?: Record, +): void { + logger.warn( + { + message_id: messageId, + status: currentAnalysis.status, + flags: currentAnalysis.flags, + score: currentAnalysis.score, + reason, + context, + timestamp: Date.now(), + }, + `Potential false positive detected: ${reason}`, + ); +} + +/** + * Log model version change. + */ +export function logModelVersionChange( + oldVersion: string, + newVersion: string, + reason: string, +): void { + logger.info( + { + old_version: oldVersion, + new_version: newVersion, + reason, + timestamp: Date.now(), + }, + `Model version changed: ${oldVersion} → ${newVersion} (${reason})`, + ); +} + +/** + * Log cache invalidation event. + */ +export function logCacheInvalidation( + source: string, + oldVersion: string, + newVersion: string, + reason: string, + affectedCount?: number, +): void { + logger.info( + { + source, + old_version: oldVersion, + new_version: newVersion, + reason, + affected_count: affectedCount, + timestamp: Date.now(), + }, + `Cache invalidation: ${source} entries with version ${oldVersion} will be ignored due to ${reason}`, + ); +} diff --git a/services/discord-gateway/src/modules/ai-moderation/textCacheStore.ts b/services/discord-gateway/src/modules/ai-moderation/textCacheStore.ts index 28cd8b4..d218b86 100644 --- a/services/discord-gateway/src/modules/ai-moderation/textCacheStore.ts +++ b/services/discord-gateway/src/modules/ai-moderation/textCacheStore.ts @@ -1,9 +1,84 @@ import { createHash } from "node:crypto"; +import { execFileSync } from "node:child_process"; import { executeAll, executeGet } from "../../shared/database/drizzle.js"; import { createChildLogger } from "../../shared/logger/logger.js"; +import { + logModelVersionChange, + logCacheInvalidation, +} from "./responseLogger.js"; const logger = createChildLogger("text-cache-store"); +/** + * Model version for vision/LLM cache entries. + * Dynamically derived from git branch name to ensure version control. + * + * Branch naming convention: + * - main/master → "main" or "master" (stable, original cache) + * - feature/terminal-fix → "feature-terminal-fix" (feature branch cache) + * - hotfix/gambling-false-positive → "hotfix-gambling-false-positive" (hotfix branch cache) + * + * Old cache entries with mismatched versions are automatically ignored. + * This ensures each branch/deployment gets fresh analyses if it changes moderation logic. + * + * Fallback: If git branch detection fails, uses environment variable or defaults to "v1". + */ +function getVisionModelVersion(): string { + try { + // Get current git branch name using execFileSync (safe, no shell injection) + const branch = execFileSync("git", ["rev-parse", "--abbrev-ref", "HEAD"], { + encoding: "utf-8", + }) + .trim() + .toLowerCase(); + + if (!branch || branch === "head") { + // Detached HEAD state — use commit hash prefix + const commit = execFileSync("git", ["rev-parse", "--short", "HEAD"], { + encoding: "utf-8", + }) + .trim() + .toLowerCase(); + return `commit-${commit}`; + } + + // Normalize branch name: replace slashes with hyphens, remove special chars + const normalized = branch + .replace(/[^a-z0-9-]/g, "-") // Replace non-alphanumeric with hyphens + .replace(/-+/g, "-") // Collapse multiple hyphens + .replace(/^-|-$/g, ""); // Remove leading/trailing hyphens + + return normalized || "v1"; + } catch (error) { + // Fallback to environment variable if git fails + const envVersion = process.env.CACHE_MODEL_VERSION; + if (envVersion) { + logger.info({ version: envVersion }, "Using CACHE_MODEL_VERSION from env"); + return envVersion; + } + + // Last resort: use stable default + logger.warn( + { + error: error instanceof Error ? error.message : String(error), + }, + "Failed to detect git branch for cache version, using fallback 'v1'", + ); + return "v1"; + } +} + +/** + * Get the current vision model version (cached at module load time). + * Version is derived from git branch name and remains constant for this process. + */ +export const VISION_MODEL_VERSION = getVisionModelVersion(); + +logger.info({ version: VISION_MODEL_VERSION }, "Vision model version initialized"); + +// Track version for invalidation logging +let previousVersion = VISION_MODEL_VERSION; + export interface TextCacheEntry { text: string; flags: string[]; @@ -15,17 +90,17 @@ export interface TextCacheEntry { /** * Lookup cached analysis result for a normalized text string. - * Returns null if not found or expired. + * Returns null if not found, expired, or model version mismatch. */ export async function getCachedText( text: string, ): Promise { try { const row = await executeGet( - `SELECT text, flags, source, analyzed_at, expires_at, hit_count + `SELECT text, flags, source, analyzed_at, expires_at, hit_count, model_version FROM text_analysis_cache - WHERE text = $1 AND expires_at > $2`, - [text, Date.now()], + WHERE text = $1 AND expires_at > $2 AND model_version = $3`, + [text, Date.now(), VISION_MODEL_VERSION], ); if (!row) return null; @@ -48,7 +123,7 @@ export async function getCachedText( } /** - * Insert or update a text analysis cache entry. + * Insert or update a text analysis cache entry with model version. */ export async function upsertCachedText( text: string, @@ -60,14 +135,15 @@ export async function upsertCachedText( try { await executeAll( - `INSERT INTO text_analysis_cache (text, flags, source, analyzed_at, expires_at, hit_count) - VALUES ($1, $2, $3, $4, $5, 0) + `INSERT INTO text_analysis_cache (text, flags, source, analyzed_at, expires_at, hit_count, model_version) + VALUES ($1, $2, $3, $4, $5, 0, $6) ON CONFLICT (text) DO UPDATE SET flags = EXCLUDED.flags, source = EXCLUDED.source, analyzed_at = EXCLUDED.analyzed_at, - expires_at = EXCLUDED.expires_at`, - [text, JSON.stringify(flags), source, now, expiresAt], + expires_at = EXCLUDED.expires_at, + model_version = EXCLUDED.model_version`, + [text, JSON.stringify(flags), source, now, expiresAt, VISION_MODEL_VERSION], ); } catch (error) { logger.error( @@ -183,17 +259,17 @@ export function makeImageCacheKey(dataUrl: string): string { /** * Lookup a cached media analysis result. - * Returns the full cached text (the analysis summary string) or null. + * Returns the full cached text (the analysis summary string) or null if not found, expired, or version mismatch. */ export async function getCachedMediaAnalysis( cacheKey: string, ): Promise { try { const row = await executeGet( - `SELECT flags, hit_count + `SELECT flags, hit_count, model_version FROM text_analysis_cache - WHERE text = $1 AND expires_at > $2`, - [cacheKey, Date.now()], + WHERE text = $1 AND expires_at > $2 AND model_version = $3`, + [cacheKey, Date.now(), VISION_MODEL_VERSION], ); if (!row) return null; @@ -211,7 +287,7 @@ export async function getCachedMediaAnalysis( } /** - * Store a media analysis result in the cache. + * Store a media analysis result in the cache with model version tracking. */ export async function upsertCachedMediaAnalysis( cacheKey: string, @@ -223,14 +299,15 @@ export async function upsertCachedMediaAnalysis( try { await executeAll( - `INSERT INTO text_analysis_cache (text, flags, source, analyzed_at, expires_at, hit_count) - VALUES ($1, $2, $3, $4, $5, 0) + `INSERT INTO text_analysis_cache (text, flags, source, analyzed_at, expires_at, hit_count, model_version) + VALUES ($1, $2, $3, $4, $5, 0, $6) ON CONFLICT (text) DO UPDATE SET flags = EXCLUDED.flags, source = EXCLUDED.source, analyzed_at = EXCLUDED.analyzed_at, - expires_at = EXCLUDED.expires_at`, - [cacheKey, JSON.stringify(analysisResult), source, now, expiresAt], + expires_at = EXCLUDED.expires_at, + model_version = EXCLUDED.model_version`, + [cacheKey, JSON.stringify(analysisResult), source, now, expiresAt, VISION_MODEL_VERSION], ); } catch (error) { logger.error( diff --git a/services/discord-gateway/src/shared/database/migrations/add_model_version_to_cache.sql b/services/discord-gateway/src/shared/database/migrations/add_model_version_to_cache.sql new file mode 100644 index 0000000..40d1387 --- /dev/null +++ b/services/discord-gateway/src/shared/database/migrations/add_model_version_to_cache.sql @@ -0,0 +1,24 @@ +-- Migration: Add model_version column to text_analysis_cache table +-- Purpose: Track which vision/LLM model version produced each cache entry +-- Reason: Invalidate old cache entries when model prompts change (e.g., terminal screenshot false positive fix) +-- Date: 2026-06-02 + +BEGIN; + +-- Add model_version column with default value +ALTER TABLE text_analysis_cache +ADD COLUMN model_version VARCHAR(50) NOT NULL DEFAULT 'v1'; + +-- Create index for efficient filtering by model version +CREATE INDEX idx_text_analysis_cache_model_version +ON text_analysis_cache(model_version); + +-- Create composite index for source + model_version queries (common pattern) +CREATE INDEX idx_text_analysis_cache_source_model_version +ON text_analysis_cache(source, model_version); + +-- Optional: Clean up old vision_llm entries that may have stale/incorrect analysis +-- Uncomment to remove all old vision analysis cache on deployment: +-- DELETE FROM text_analysis_cache WHERE source = 'vision_llm' AND model_version = 'v1'; + +COMMIT; diff --git a/services/discord-gateway/src/shared/database/schema.ts b/services/discord-gateway/src/shared/database/schema.ts index d122471..d935bba 100644 --- a/services/discord-gateway/src/shared/database/schema.ts +++ b/services/discord-gateway/src/shared/database/schema.ts @@ -361,6 +361,9 @@ export const pgRetentionPoliciesTable = pgTable( * * Uses the FULL normalized text (not per-word) because context matters: * "kau" alone is clean, but "awas kau" can be a threat. + * + * model_version tracks the vision/LLM model version that produced this cache entry. + * On model updates, bump the version to invalidate all old cache entries automatically. */ export const pgTextAnalysisCacheTable = pgTable( "text_analysis_cache", @@ -381,12 +384,20 @@ export const pgTextAnalysisCacheTable = pgTable( expires_at: pgBigint("expires_at", { mode: "number" }).notNull(), /** How many times this cached text has been reused. */ hit_count: pgInteger("hit_count").notNull().default(0), + /** Model version that produced this cache entry (e.g. "v1", "v2-2026-06-02"). Mismatched versions are ignored. */ + model_version: pgText("model_version").notNull().default("v1"), }, (table) => ({ expiresAtIdx: pgIndex("idx_text_analysis_cache_expires_at").on( table.expires_at, ), sourceIdx: pgIndex("idx_text_analysis_cache_source").on(table.source), + modelVersionIdx: pgIndex("idx_text_analysis_cache_model_version").on( + table.model_version, + ), + sourceModelVersionIdx: pgIndex( + "idx_text_analysis_cache_source_model_version", + ).on(table.source, table.model_version), }), );