feat: add response logger for moderation/vision audit trail + cache model versioning

2026-06-02 12:25:12 +07:00
parent f4db42e11c 385fc0bbbf
commit e4302c22b4
6 changed files with 505 additions and 18 deletions
@@ -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<N>" or "v<N>-<date>-<description>"
+# Example progression: v1 → v2-2026-06-02-terminal-fix → v3-2026-06-15-new-model
+# CACHE_MODEL_VERSION=v2-2026-06-02
+
@@ -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);

@@ -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<string, any>,
+): 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<string, any>,
+): 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<string, number>,
+): 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<string, any>,
+): 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}`,
+  );
+}
@@ -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<TextCacheEntry | null> {
  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<string | null> {
  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(
@@ -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;
@@ -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),
  }),
 );