fix(ai-moderation): matikan hot requeue loop saat upload attachment in-flight

Batch race guard balikin {ok:true, rows:[]} tanpa sinyal saat semua target
masih upload-pending -> processor klasifikasi semua incomplete -> fanout ke
individual queue -> di situ requeue + reschedule 250ms -> balik ke batch:
hot loop ~300ms sepanjang upload (10 siklus/3 dtk di log prod 08:13).

Fix: worker batch kini return uploadPendingIds eksplisit; classifier pure
baru (partitionBatchOutcome) partisi completed/upload_pending/incomplete/
parse_failed/api_failed; target upload-pending DEFERRED dengan poll backoff
linear (AI_ANALYSIS_UPLOAD_POLL_MS 1500 base, cap AI_ANALYSIS_MAX_UPLOAD_POLL_MS
8000), tidak pernah masuk fanout; tail shouldScheduleNext tak menimpa defer.
Test: tests/batchOutcomeClassifier.test.ts (8 kasus, pure tanpa DB/Piscina).

services/discord-gateway/src/modules/ai-moderation/ai-analysis-worker.ts

@@ -93,6 +93,12 @@ type BatchOkResponse = {
   ok: true;
   conversationKey: string;
   rows: MessageRecord[];
+  /**
+   * Race-guard signal (2026-08-25): target ids whose attachment upload is
+   * still in-flight — NO analysis ran for them. The processor must defer
+   * these (requeue + poll), never fan them out as failures.
+   */
+  uploadPendingIds?: string[];
 };
 type BatchErrorResponse = {
   ok: false;
@@ -319,7 +325,16 @@ async function processBatch(job: {
       ? messages
       : messages.filter((m) => !pendingUploadTargetIds.has(m.id));
   if (readyMessages.length === 0) {
-    return { ok: true, conversationKey, rows: [] };
+    // Explicit signal (2026-08-25): every target is still upload-pending.
+    // Returning bare {ok:true, rows:[]} made the processor classify all of
+    // them "incomplete" and fan out to the individual queue — a hot ~300ms
+    // requeue loop for the whole upload duration.
+    return {
+      ok: true,
+      conversationKey,
+      rows: [],
+      uploadPendingIds: messages.map((m) => m.id),
+    };
   }
 
   // The orchestrator handles text/media split + caching + parallel paths

services/discord-gateway/src/modules/ai-moderation/batchOutcomeClassifier.ts

@@ -0,0 +1,108 @@
+/**
+ * batchOutcomeClassifier.ts
+ *
+ * Pure partitioner of the batch worker response (2026-08-25).
+ *
+ * Bug history: the batch race guard returned `{ok:true, rows:[]}` when every
+ * target's attachment upload was still in-flight. The processor classified all
+ * of them as "incomplete" and fanned out to the individual queue, where the
+ * guard there requeued + rescheduled at the 250ms debounce — a hot ~300ms loop
+ * for the entire upload duration (~10 cycles in 3s in prod logs). Root fix:
+ * the worker now reports `uploadPendingIds` explicitly and this pure function
+ * partitions the outcome so upload-pending targets NEVER enter the fanout.
+ */
+
+export interface BatchRowLike {
+  id?: string;
+  ai_status?: string | null;
+  ai_moderation_flags?: string | null;
+}
+
+export interface BatchWorkerResponseLike {
+  ok?: boolean;
+  rows?: BatchRowLike[];
+  /** Explicit race-guard signal from the worker (2026-08-25). */
+  uploadPendingIds?: string[];
+  error?: string;
+}
+
+/** One target's per-message disposition after a batch attempt. */
+export type BatchTargetKind =
+  | "completed"
+  | "upload_pending"
+  | "incomplete"
+  | "parse_failed"
+  | "api_failed";
+
+function flagsOf(row: { ai_moderation_flags?: string | null }): string[] {
+  if (!row.ai_moderation_flags) return [];
+  try {
+    const parsed = JSON.parse(row.ai_moderation_flags) as unknown;
+    return Array.isArray(parsed) ? (parsed as string[]) : [];
+  } catch {
+    return [] as string[];
+  }
+}
+
+/**
+ * Partition the input message ids into per-message dispositions for one batch
+ * worker response. Pure: no DB/Piscina/logger — unit-testable directly.
+ *
+ * Priority per id: explicit uploadPendingIds → completed row → flag-based
+ * failure kinds → unexplained missing (treated like incomplete).
+ */
+export function partitionBatchOutcome(
+  messages: ReadonlyArray<{ id: string }>,
+  response: BatchWorkerResponseLike,
+): Map<string, BatchTargetKind> {
+  const pendingSet = new Set(response.uploadPendingIds ?? []);
+  const rowsById = new Map(
+    (response.rows ?? [])
+      .filter((r): r is BatchRowLike & { id: string } => Boolean(r?.id))
+      .map((r) => [r.id, r]),
+  );
+
+  const out = new Map<string, BatchTargetKind>();
+  for (const msg of messages) {
+    if (pendingSet.has(msg.id)) {
+      out.set(msg.id, "upload_pending");
+      continue;
+    }
+    const row = rowsById.get(msg.id);
+    if (!row) {
+      // Unexplained drop: LLM silently omitted it. Same retryable bucket as
+      // analysis_incomplete — never a silent success.
+      out.set(msg.id, "incomplete");
+      continue;
+    }
+    if (row.ai_status !== "error") {
+      out.set(msg.id, "completed");
+      continue;
+    }
+    const flags = flagsOf(row);
+    if (flags.includes("analysis_incomplete")) {
+      out.set(msg.id, "incomplete");
+    } else if (flags.includes("analysis_parse_failed")) {
+      out.set(msg.id, "parse_failed");
+    } else if (flags.includes("analysis_api_failed")) {
+      out.set(msg.id, "api_failed");
+    } else {
+      out.set(msg.id, "incomplete");
+    }
+  }
+  return out;
+}
+
+/**
+ * Linear backoff ramp for consecutive upload-pending polls:
+ * poll N (1-based) waits min(base × N, cap). Keeps latency low for fast
+ * uploads while bounding total polling cost for long uploads.
+ */
+export function computeUploadPollDelayMs(
+  consecutivePolls: number,
+  baseMs: number,
+  capMs: number,
+): number {
+  const n = Math.max(1, Math.floor(consecutivePolls));
+  return Math.min(Math.round(baseMs * n), Math.round(capMs));
+}

services/discord-gateway/src/modules/ai-moderation/batchProcessor.ts

@@ -4,6 +4,10 @@ import { isAgeRestrictedMetadata } from "../message-capture/messageMetadata.js";
 import { messageStore } from "../message-capture/messageStore.js";
 import type { MessageRecord } from "../message-capture/types.js";
 import { pickBatchWithinBudget as pickBatchWithinBudgetPure } from "./batchBudget.js";
+import {
+  computeUploadPollDelayMs,
+  partitionBatchOutcome,
+} from "./batchOutcomeClassifier.js";
 import { workerPool } from "./circuitBreaker.js";
 import { estimateTokens } from "./conversationContext.js";
 import {
@@ -21,11 +25,20 @@ import {
 
 const logger = createChildLogger("batch-processor");
 
+/**
+ * Consecutive upload-pending poll counter per conversation (2026-08-25).
+ * Drives the linear backoff ramp while attachments are still uploading;
+ * cleared as soon as a batch comes back with no upload-pending targets.
+ */
+const conversationUploadPolls = new Map<string, number>();
+
 export interface AnalysisWorkerResponse {
   ok: boolean;
   conversationKey: string;
   rows: MessageRecord[];
   error?: string;
+  /** Explicit upload-in-flight signal from the batch race guard (2026-08-25). */
+  uploadPendingIds?: string[];
 }
 
 // ---------------------------------------------------------------------------
@@ -141,6 +154,8 @@ export async function processBatch(
 
   activeRequests++;
   let shouldScheduleNext = false;
+  /** Set when upload-pending targets defer the next cycle by this many ms. */
+  let deferredUploadRescheduleMs: number | null = null;
   try {
     const result = (await workerPool.run({
       type: "batch",
@@ -199,37 +214,85 @@ export async function processBatch(
       return;
     }
 
-    // Batch succeeded -- check for messages the LLM silently dropped or failed
-    const incompleteMessages: MessageRecord[] = [];
-    const parseFailedMessages: MessageRecord[] = [];
+    // Batch succeeded -- partition per-message outcome explicitly (2026-08-25).
+    // upload_pending targets are DEFERRED (never fanned out): the old code
+    // treated them as incomplete -> individual queue -> requeue+250ms
+    // reschedule -> hot ~300ms loop for the whole upload duration.
+    const outcomeById = partitionBatchOutcome(messages, result);
+    const messagesForIndividualQueue: MessageRecord[] = [];
     const apiFailedMessages: MessageRecord[] = [];
+    const uploadPendingMessages: MessageRecord[] = [];
 
     for (const msg of messages) {
-      const row = result.rows.find((r) => r.id === msg.id);
-      if (!row) {
-        incompleteMessages.push(msg);
-        continue;
-      }
-      if (row.ai_status === "error") {
-        let flags: string[] = [];
-        try {
-          flags = JSON.parse(row.ai_moderation_flags ?? "[]") as string[];
-        } catch {}
-
-        if (flags.includes("analysis_incomplete")) {
-          incompleteMessages.push(msg);
-        } else if (flags.includes("analysis_parse_failed")) {
-          parseFailedMessages.push(msg);
-        } else if (flags.includes("analysis_api_failed")) {
+      switch (outcomeById.get(msg.id)) {
+        case "upload_pending":
+          uploadPendingMessages.push(msg);
+          break;
+        case "api_failed":
+          // Preserve the dedicated api-failure semantics below: revert +
+          // conversation cooldown instead of an immediate individual retry.
           apiFailedMessages.push(msg);
-        }
+          break;
+        default:
+          // incomplete / parse_failed / unexplained drops stay retryable via
+          // the individual fallback queue (same semantics as before).
+          messagesForIndividualQueue.push(msg);
+          break;
       }
     }
 
-    const messagesForIndividualQueue = [
-      ...incompleteMessages,
-      ...parseFailedMessages,
-    ];
+    if (uploadPendingMessages.length > 0) {
+      const polls = (conversationUploadPolls.get(conversationKey) ?? 0) + 1;
+      conversationUploadPolls.set(conversationKey, polls);
+      const delayMs = computeUploadPollDelayMs(
+        polls,
+        config.AI_ANALYSIS_UPLOAD_POLL_MS,
+        config.AI_ANALYSIS_MAX_UPLOAD_POLL_MS,
+      );
+      logger.debug(
+        {
+          conversationKey,
+          count: uploadPendingMessages.length,
+          ids: uploadPendingMessages.map((m) => m.id),
+          pollAttempt: polls,
+          delayMs,
+        },
+        "Attachment upload in-flight for batch targets — deferring with poll backoff",
+      );
+
+      // Put the rows back to `pending` so the scheduler owns them again.
+      await messageStore
+        .updateMessagesAIAnalysisBulk(
+          uploadPendingMessages.map((msg) => ({
+            messageId: msg.id,
+            result: {
+              status: "pending",
+              flags: null,
+              score: null,
+              analysis: null,
+              categories: null,
+              severity: null,
+              confidence: null,
+              recommendedAction: null,
+              analyzedAt: null,
+              error: null,
+            },
+          })),
+        )
+        .catch((err: unknown) => {
+          logger.error(
+            { error: String(err), ids: uploadPendingMessages.map((m) => m.id) },
+            "Failed to revert upload-pending batch targets to pending",
+          );
+          return [] as MessageRecord[];
+        });
+
+      // Poll backoff instead of the 250ms debounce: the finally-block
+      // schedules the next cycle after this delay instead of immediately.
+      deferredUploadRescheduleMs = delayMs;
+    } else {
+      conversationUploadPolls.delete(conversationKey);
+    }
 
     if (messagesForIndividualQueue.length > 0) {
       logger.warn(
@@ -307,7 +370,11 @@ export async function processBatch(
       resetConversationBatchFailures(conversationKey);
       conversationErrorCooldown.delete(conversationKey);
     }
-    shouldScheduleNext = true;
+    // Upload-pending defer owns the next-cycle timing; don't let the default
+    // immediate schedule override it.
+    if (deferredUploadRescheduleMs === null) {
+      shouldScheduleNext = true;
+    }
   } catch (error) {
     recordConversationBatchFailure(conversationKey);
 
@@ -344,7 +411,17 @@ export async function processBatch(
     if (conversationProcessing.get(conversationKey) === processingStartedAt) {
       conversationProcessing.delete(conversationKey);
     }
-    if (shouldScheduleNext) {
+    if (deferredUploadRescheduleMs !== null) {
+      // Upload still in-flight: re-schedule after the backoff delay instead of
+      // immediately (the old path hot-looped at ~250-300ms per cycle).
+      const delayMs = deferredUploadRescheduleMs;
+      setTimeout(() => {
+        // Dynamic import to avoid circular dependency at module scope
+        import("./batchScheduler.js").then((m) =>
+          m.scheduleConversationAnalysis(conversationKey),
+        );
+      }, delayMs).unref();
+    } else if (shouldScheduleNext) {
       setImmediate(() => {
         // Dynamic import to avoid circular dependency at module scope
         import("./batchScheduler.js").then((m) =>

services/discord-gateway/src/shared/config/index.ts

@@ -254,6 +254,12 @@ export const configSchema = z
       .positive()
       .default(10000),
     AI_ANALYSIS_ERROR_COOLDOWN_MS: z.coerce.number().positive().default(30000),
+    // Upload-pending batch poll (2026-08-25): when a batch is deferred because
+    // attachments are still uploading, the processor re-schedules with this
+    // base delay (linear ramp per consecutive poll, capped) instead of the
+    // 250ms debounce — the old path hot-looped ~300ms for the whole upload.
+    AI_ANALYSIS_UPLOAD_POLL_MS: z.coerce.number().positive().default(1500),
+    AI_ANALYSIS_MAX_UPLOAD_POLL_MS: z.coerce.number().positive().default(8000),
 
     // ── AI Analysis Batch ───────────────────────────────────────────────
     AI_ANALYSIS_MAX_BATCH_SIZE: z.coerce.number().int().positive().default(200),

services/discord-gateway/tests/batchOutcomeClassifier.test.ts

@@ -0,0 +1,110 @@
+// ═══════════════════════════════════════════════════════════════════════════
+// partitionBatchOutcome — upload-pending defer vs fanout (2026-08-25)
+// ═══════════════════════════════════════════════════════════════════════════
+// Bug history: the batch worker's race guard returned {ok:true, rows:[]} while
+// attachments were still uploading; every target was classified "incomplete",
+// fanned out to the individual queue, requeued there, rescheduled at 250ms —
+// a hot ~300ms loop for the whole upload duration (~10 cycles in 3s in prod).
+import { describe, expect, it } from "vitest";
+import {
+  computeUploadPollDelayMs,
+  partitionBatchOutcome,
+} from "../src/modules/ai-moderation/batchOutcomeClassifier.js";
+
+const msgs = (...ids: string[]) => ids.map((id) => ({ id }));
+
+describe("partitionBatchOutcome", () => {
+  it("marks ALL targets upload_pending when the full-batch guard fires", () => {
+    const out = partitionBatchOutcome(msgs("a", "b"), {
+      ok: true,
+      rows: [],
+      uploadPendingIds: ["a", "b"],
+    });
+    expect(out.get("a")).toBe("upload_pending");
+    expect(out.get("b")).toBe("upload_pending");
+  });
+
+  it("never classifies an explicit upload_pending id as incomplete", () => {
+    // The regression this file exists for: uploadPendingIds must win over the
+    // missing-row heuristic.
+    const out = partitionBatchOutcome(msgs("a"), {
+      ok: true,
+      rows: [],
+      uploadPendingIds: ["a"],
+    });
+    expect(out.get("a")).not.toBe("incomplete");
+    expect(out.get("a")).toBe("upload_pending");
+  });
+
+  it("partitions a mixed batch: completed + upload-pending + missing", () => {
+    const out = partitionBatchOutcome(msgs("ok1", "up1", "gone1"), {
+      ok: true,
+      rows: [
+        { id: "ok1", ai_status: "clean" },
+        // up1 has NO row but IS in uploadPendingIds -> deferred, not failed
+      ],
+      uploadPendingIds: ["up1"],
+    });
+    expect(out.get("ok1")).toBe("completed");
+    expect(out.get("up1")).toBe("upload_pending");
+    expect(out.get("gone1")).toBe("incomplete"); // unexplained drop stays retryable
+  });
+
+  it("routes flag-based failures to their buckets", () => {
+    const out = partitionBatchOutcome(msgs("i", "p", "f"), {
+      ok: true,
+      rows: [
+        {
+          id: "i",
+          ai_status: "error",
+          ai_moderation_flags: JSON.stringify(["analysis_incomplete"]),
+        },
+        {
+          id: "p",
+          ai_status: "error",
+          ai_moderation_flags: JSON.stringify(["analysis_parse_failed"]),
+        },
+        {
+          id: "f",
+          ai_status: "error",
+          ai_moderation_flags: JSON.stringify(["analysis_api_failed"]),
+        },
+      ],
+    });
+    expect(out.get("i")).toBe("incomplete");
+    expect(out.get("p")).toBe("parse_failed");
+    expect(out.get("f")).toBe("api_failed");
+  });
+
+  it("treats error rows with no known flag as retryable incomplete", () => {
+    const out = partitionBatchOutcome(msgs("x"), {
+      ok: true,
+      rows: [
+        { id: "x", ai_status: "error", ai_moderation_flags: '["weird_flag"]' },
+      ],
+    });
+    expect(out.get("x")).toBe("incomplete");
+  });
+
+  it("accepts DB-round-trip MessageRecord shape (null ai_status)", () => {
+    const out = partitionBatchOutcome([{ id: "r", ai_status: null }] as never, {
+      ok: true,
+      rows: [{ id: "r", ai_status: null }],
+    });
+    expect(out.get("r")).toBe("completed");
+  });
+});
+
+describe("computeUploadPollDelayMs", () => {
+  it("ramps linearly and respects the cap", () => {
+    expect(computeUploadPollDelayMs(1, 1500, 8000)).toBe(1500);
+    expect(computeUploadPollDelayMs(2, 1500, 8000)).toBe(3000);
+    expect(computeUploadPollDelayMs(3, 1500, 8000)).toBe(4500);
+    expect(computeUploadPollDelayMs(9, 1500, 8000)).toBe(8000); // capped
+  });
+
+  it("is safe on degenerate input", () => {
+    expect(computeUploadPollDelayMs(0, 1500, 8000)).toBe(1500);
+    expect(computeUploadPollDelayMs(-5, 1000, 4000)).toBe(1000);
+  });
+});