fix(moderation): patch 6 aiAnalyzer audit vulnerabilities

#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<conversationKey, count>
  - 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

@@ -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"),

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<string, NodeJS.Timeout>();
-// Track conversations currently being processed
-const conversationProcessing = new Map<string, number>();
-// Track conversations in error cooldown (failed recently)
-const conversationErrorCooldown = new Map<string, number>();
+// ---------------------------------------------------------------------------
+// Batch pipeline state
+// ---------------------------------------------------------------------------
 
-const AI_PROCESSING_OVERLAP_MS = 30000;
+/** Debounce timer handle per conversation key. */
+const conversationDebounceTimers = new Map<string, NodeJS.Timeout>();
+/** Timestamp of when processing started per conversation key. */
+const conversationProcessing = new Map<string, number>();
+/** Cooldown expiry timestamp per conversation key after an error. */
+const conversationErrorCooldown = new Map<string, number>();
 
 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<messageId> de-duplicates so the same message can't be
+//    in-flight twice (Discord snowflakes are globally unique, but be safe).
+//  • A Map<conversationKey, count> 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<string>();
 
-/** 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<string, number>();
+
+/** 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<void> {
   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<void> {
   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<void> {
   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<void> {
 }
 
 /**
- * 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<void>[] = [];
+          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);
 }

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<string[]> {
+  try {
+    const database = db();
+    const rows = await database
+      .selectDistinct<Array<{ thread_id: string | null; channel_id: string }>>({
+        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<MessageRecord[]> {
+  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;
+  }
+}

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;
 }