diff --git a/.hermes/plans/2026-08-24-attachment-delay-fix.md b/.hermes/plans/2026-08-24-attachment-delay-fix.md new file mode 100644 index 00000000..8a673e34 --- /dev/null +++ b/.hermes/plans/2026-08-24-attachment-delay-fix.md @@ -0,0 +1,56 @@ +# Spec: Perbaiki Delay Attachment 162s→<20s (GMW AI Analysis) + +Tanggal: 2026-08-24 · Repo `~/GMW` · Service discord-gateway + +## Evidence (audit produksi) + +Klaster pesan attachment delay ~330–400 detik. Trace pesan `1541417073245290638` (.gif): +19:01:08 dibuat → 19:01:09 batch incomplete → fan-out individual → **guard upload-pending +mengembalikan `results:[]`** → diperalakukan sukses (`complete ... (undefined)`) → row +tertahan `ai_status='processing'` **tanpa penanggung jawab** → 19:06:12 cleanup mengembalikan +ke `pending` (tepat 300s) → baru dianalisis. Plus vision gagal 3× utk GIF besar +("Stream ended before producing a non-ping SSE event") → degradasi teks. + +## Root causes + +- **A (fatal)**: `individualFallbackProcessor.processIndividualFallback` memperlakukan + `ok:true + results:[]` sebagai sukses. Race-guard upload di `ai-analysis-worker.processIndividual` + sengaja balik `results:[]` (desain lama) → pesan yatim `processing` sampai cleanup 300s. +- **B**: `llmVision` hanya mencoba `stream:true`; kegagalan SSE truncation pada gambar besar + = 3 retry sia-sia (semua jalur sama) → bukti media hilang. +- **C**: safety-net cleanup 300s terlalu lambat sbg satu-satunya pemulih `processing`. + +## Fix + +1. **F1 — sinyal eksplisit upload-pending**: `IndividualOkResponse` + field opsional + `uploadPending?: boolean`. Worker set `uploadPending:true` saat race guard kena. +2. **F2 — processor menangani 3 kondisi** via helper murni baru + `classifyIndividualWorkerResult(result): "success" | "upload_pending" | "incomplete" | "error"` + (modul baru `fallbackResultClassifier.ts`, zero-dep agar mudah dites): + - `upload_pending` → tulis ulang row ke `pending` (pola sama dgn revert apiFailed di + batchProcessor) + broadcast + **re-schedule analisis percakapan segera** + (dynamic import batchScheduler, pola anti-siklus yg sudah ada) → retry dalam ~250ms + begitu upload beres. Bukan error, tidak naikkan CB counter. + - `incomplete` (flags analysis_incomplete) → perilaku lama (exhausted path). + - `error` / `results kosong tanpa penjelasan` → throw transien (retry oleh recovery), + BUKAN sukses palsu. Log "(undefined)" hilang. +3. **F3 — vision non-stream fallback**: di `llmVision`, jika error match + `/Stream ended before producing a non-ping SSE|stream ended/i` → coba SEKALI lagi dengan + `stream:false` (router agregasi penuh; timeout tetap 60s). Konversi hard-fail jadi sukses. +4. **F4 — turunkan safety net**: default `revertStuckProcessingMessages` 300000 → 120000 ms. + +## File disentuh + +- `src/modules/ai-moderation/fallbackResultClassifier.ts` (BARU, pure) +- `src/modules/ai-moderation/ai-analysis-worker.ts` (tipe + set flag uploadPending) +- `src/modules/ai-moderation/individualFallbackProcessor.ts` (konsumsi classifier + reschedule) +- `src/modules/ai-moderation/llmClient.ts` (fallback non-stream di llmVision) +- `src/modules/message-capture/messagesCleanup.ts` (default 120s) + +## Verifikasi + +- Test baru `tests/fallbackResultClassifier.test.ts` (4 klasifikasi + edge kosong). +- Gate: tsc --noEmit, biome error-level, vitest run semua hijau. +- Deploy GHA sukses; pasca-deploy: pesan attachment baru p50 < 20s + (`SELECT percentile_cont(0.5) ... WHERE metadata attachments>0 AND created_at > deploy`), + tidak ada lagi "complete ... (undefined)". diff --git a/services/discord-gateway/src/modules/ai-moderation/ai-analysis-worker.ts b/services/discord-gateway/src/modules/ai-moderation/ai-analysis-worker.ts index ce80ae25..c29e61de 100644 --- a/services/discord-gateway/src/modules/ai-moderation/ai-analysis-worker.ts +++ b/services/discord-gateway/src/modules/ai-moderation/ai-analysis-worker.ts @@ -100,7 +100,16 @@ type BatchErrorResponse = { rows: MessageRecord[]; error: string; }; -type IndividualOkResponse = { ok: true; results: AnalysisResult[] }; +type IndividualOkResponse = { + ok: true; + results: AnalysisResult[]; + /** + * Race-guard signal (2026-08-24): the message's attachment upload is still + * in-flight — NO analysis ran. The processor must re-queue the message as + * `pending` and re-schedule, never treat this as a completed moderation. + */ + uploadPending?: boolean; +}; type IndividualErrorResponse = { ok: false; results: AnalysisResult[]; @@ -415,7 +424,7 @@ async function processIndividual(job: { (a) => a.message_id === message.id && a.upload_status === "pending", ); if (uploadStillPending) { - return { ok: true, results: [] }; + return { ok: true, results: [], uploadPending: true }; } try { diff --git a/services/discord-gateway/src/modules/ai-moderation/fallbackResultClassifier.ts b/services/discord-gateway/src/modules/ai-moderation/fallbackResultClassifier.ts new file mode 100644 index 00000000..988c9bff --- /dev/null +++ b/services/discord-gateway/src/modules/ai-moderation/fallbackResultClassifier.ts @@ -0,0 +1,65 @@ +/** + * fallbackResultClassifier.ts + * + * Pure classifier for the individual-fallback worker response. + * + * Bug history (2026-08-24): the worker's upload-pending race guard returned + * `{ ok: true, results: [] }` (a legacy "no results yet" signal), but the + * processor treated ANY `ok:true` as a successful moderation. Empty results + * meant nothing was written to the DB — the message stayed stuck in + * `ai_status='processing'` with nobody watching it until the 300s cleanup + * reverted it. That single gap produced the ~330-400s attachment delay + * cluster. Classification now happens in ONE pure function so every outcome + * has an explicit, testable owner. + */ + +export type WorkerResultKind = + | "success" + | "upload_pending" + | "incomplete" + | "error"; + +export interface ClassifiableWorkerResult { + ok?: boolean; + /** Upload-pending marker set by ai-analysis-worker's race guard. */ + uploadPending?: boolean; + results?: Array<{ status?: string; flags?: string[] | string } | undefined>; + error?: string; +} + +function flagsOf(r: { flags?: string[] | string }): string[] { + if (!r.flags) return []; + if (Array.isArray(r.flags)) return r.flags; + try { + const parsed = JSON.parse(r.flags) as unknown; + return Array.isArray(parsed) ? (parsed as string[]) : []; + } catch { + return []; + } +} + +/** + * Classify an individual-fallback worker response: + * - "upload_pending": explicit race-guard signal — retry shortly, NOT an error. + * - "success": at least one result and none is analysis_incomplete. + * - "incomplete": LLM ran but dropped/failed this message after retries + * (analysis_incomplete flag) — terminal exhausted path. + * - "error": anything else (ok:false, or ok:true with NO explainable + * results). The old code silently succeeded here — never again. + */ +export function classifyIndividualWorkerResult( + result: ClassifiableWorkerResult, +): WorkerResultKind { + if (result.uploadPending === true) return "upload_pending"; + const results = (result.results ?? []).filter( + (r): r is NonNullable => Boolean(r), + ); + if (results.length === 0) return "error"; + if (result.ok !== true) return "error"; + for (const r of results) { + const flags = flagsOf(r); + if (flags.includes("analysis_incomplete")) return "incomplete"; + if ((r.status ?? "") === "") return "error"; + } + return "success"; +} diff --git a/services/discord-gateway/src/modules/ai-moderation/individualFallbackProcessor.ts b/services/discord-gateway/src/modules/ai-moderation/individualFallbackProcessor.ts index df32c694..5e186183 100644 --- a/services/discord-gateway/src/modules/ai-moderation/individualFallbackProcessor.ts +++ b/services/discord-gateway/src/modules/ai-moderation/individualFallbackProcessor.ts @@ -8,6 +8,7 @@ import type { } from "../message-capture/types.js"; import { getConversationKey, workerPool } from "./circuitBreaker.js"; import { fireAlert } from "./conversationState.js"; +import { classifyIndividualWorkerResult } from "./fallbackResultClassifier.js"; import { broadcastAnalysisCompleted, LAST_ERROR, @@ -78,26 +79,82 @@ async function processIndividualFallback( message, skipNormalAnalysis: false, } as unknown)) as - | { ok: true; results: AnalysisResult[] } + | { ok: true; results: AnalysisResult[]; uploadPending?: boolean } | { ok: false; results: AnalysisResult[]; error: string }; + // Explicit outcome classification (2026-08-24): the old code treated any + // ok:true as a completed moderation, so the upload-pending race guard's + // empty results left messages stuck in `processing` until the 300s + // cleanup reverted them — the root cause of the ~330s attachment delays. + const kind = classifyIndividualWorkerResult(workerResult); + + if (kind === "upload_pending") { + // Attachment still uploading — put the row back to `pending` and + // re-schedule this conversation immediately. The next scheduler cycle + // (~debounce 250ms) re-fetches; once upload_status flips to done the + // race guard passes and analysis proceeds. NOT an error: never touches + // the circuit breaker counters. + const revertedRows = await messageStore + .updateMessagesAIAnalysisBulk([ + { + messageId, + result: { + status: "pending", + flags: null, + score: null, + analysis: null, + categories: null, + severity: null, + confidence: null, + recommendedAction: null, + analyzedAt: null, + error: null, + }, + }, + ]) + .catch((dbErr: unknown) => { + logger.error( + { messageId, error: String(dbErr) }, + "Failed to revert upload-pending message to pending", + ); + return [] as MessageRecord[]; + }); + for (const row of revertedRows) { + broadcastAnalysisCompleted(row); + } + logger.debug( + { messageId, conversationKey }, + "Individual fallback: attachment upload in-flight — requeued as pending + rescheduled", + ); + setImmediate(() => { + import("./batchScheduler.js") + .then((m) => m.scheduleConversationAnalysis(conversationKey)) + .catch(() => {}); + }); + return; + } + let analysisResult: { results: AnalysisResult[] } | null = null; - if (workerResult.ok) { - const stillIncomplete = workerResult.results.some((r) => - r.flags.includes("analysis_incomplete"), + if (kind === "success") { + analysisResult = workerResult; + } else if (kind === "incomplete") { + exhaustedOnIncomplete = true; + analysisResult = null; + } else { + // "error" — includes ok:true with unexplainable empty results (the old + // silent-success bug). Throw so it is treated as a transient failure. + throw new Error( + (workerResult as { error?: string }).error ?? + "Individual worker returned no explainable results", ); - if (stillIncomplete) { - exhaustedOnIncomplete = true; - analysisResult = null; - } else { - analysisResult = workerResult; - } } // No heuristic fallback: an incomplete/errored LLM result stays a // retryable error — the recovery worker picks it up later. Producing a // regex/wordlist verdict here would reintroduce false positives. + // (incomplete keeps its exhausted flag so the catch writes the terminal + // individual_analysis_exhausted status.) if (!analysisResult) { throw new Error(`LLM analysis failed for message ${messageId}`); } diff --git a/services/discord-gateway/src/modules/ai-moderation/llmClient.ts b/services/discord-gateway/src/modules/ai-moderation/llmClient.ts index 89e7f641..a68020c8 100644 --- a/services/discord-gateway/src/modules/ai-moderation/llmClient.ts +++ b/services/discord-gateway/src/modules/ai-moderation/llmClient.ts @@ -356,10 +356,10 @@ export async function llmVision( promptText: string, imageUrl: { url: string }, ): Promise { - const completion = await llmChat({ + const params = { messages: [ { - role: "user", + role: "user" as const, content: [ { type: "text" as const, text: promptText }, { type: "image_url" as const, image_url: imageUrl }, @@ -371,9 +371,26 @@ export async function llmVision( temperature: 0.1, top_p: 0.9, retries: 0, - stream: true, // router always streams SSE; non-stream waits for full body and times out timeout: config.AI_LLM_VISION_ANALYSIS_TIMEOUT_MS ?? 60_000, - }); + }; + + // Streaming first (the router always streams SSE; a non-stream request + // waits for the full body and times out on slow models). Fallback (2026-08-24): + // large GIFs/images sometimes get their SSE stream truncated mid-flight by + // the upstream ("Stream ended before producing a non-ping SSE event") — all + // streaming retries fail identically, so retry ONCE with stream:false where + // the router assembles the complete response server-side. + let completion; + try { + completion = await llmChat({ ...params, stream: true }); + } catch (err) { + const msg = err instanceof Error ? err.message : String(err); + if (/stream ended before producing a non-ping sse/i.test(msg)) { + completion = await llmChat({ ...params, stream: false }); + } else { + throw err; + } + } if (!completion) return null; return completion.choices[0]?.message?.content?.trim() ?? null; diff --git a/services/discord-gateway/src/modules/message-capture/messagesCleanup.ts b/services/discord-gateway/src/modules/message-capture/messagesCleanup.ts index 2ae095d1..15f23fd7 100644 --- a/services/discord-gateway/src/modules/message-capture/messagesCleanup.ts +++ b/services/discord-gateway/src/modules/message-capture/messagesCleanup.ts @@ -48,7 +48,13 @@ export class MessagesCleanup { } async revertStuckProcessingMessages( - timeoutMs: number = 300000, + // 2026-08-24: lowered from 300000 — this cleanup is the last-resort + // recoverer for rows stuck in `processing`. With the upload-pending + // race-guard now requeueing properly (fallbackResultClassifier), any row + // that still sits here for >2min is a genuine leak; reverting sooner + // bounds the worst-case delay without racing legitimate in-flight work + // (media batches can legitimately take ~60s+). + timeoutMs: number = 120000, ): Promise { this.logger.debug({ timeoutMs }, "revertStuckProcessingMessages entry"); try { diff --git a/services/discord-gateway/tests/fallbackResultClassifier.test.ts b/services/discord-gateway/tests/fallbackResultClassifier.test.ts new file mode 100644 index 00000000..e9e1ac72 --- /dev/null +++ b/services/discord-gateway/tests/fallbackResultClassifier.test.ts @@ -0,0 +1,79 @@ +// ═══════════════════════════════════════════════════════════════════════════ +// classifyIndividualWorkerResult — upload-pending vs success vs incomplete vs error +// ═══════════════════════════════════════════════════════════════════════════ +// Bug (2026-08-24): the worker's upload-pending race guard returned +// {ok:true, results:[]}; the processor treated it as a completed moderation, +// leaving messages stuck in `processing` until the 300s cleanup reverted them. +import { describe, expect, it } from "vitest"; +import { classifyIndividualWorkerResult } from "../src/modules/ai-moderation/fallbackResultClassifier.js"; + +describe("classifyIndividualWorkerResult", () => { + it("classifies the upload-pending race guard signal FIRST", () => { + expect( + classifyIndividualWorkerResult({ + ok: true, + results: [], + uploadPending: true, + }), + ).toBe("upload_pending"); + }); + + it("classifies a normal verdict as success", () => { + expect( + classifyIndividualWorkerResult({ + ok: true, + results: [{ status: "clean", flags: [] }], + }), + ).toBe("success"); + }); + + it("classifies analysis_incomplete as incomplete", () => { + expect( + classifyIndividualWorkerResult({ + ok: true, + results: [{ status: "error", flags: ["analysis_incomplete"] }], + }), + ).toBe("incomplete"); + }); + + it("accepts flags as JSON string (DB round-trip shape)", () => { + expect( + classifyIndividualWorkerResult({ + ok: true, + results: [ + { status: "error", flags: JSON.stringify(["analysis_incomplete"]) }, + ], + }), + ).toBe("incomplete"); + }); + + it("classifies ok:false as error", () => { + expect( + classifyIndividualWorkerResult({ + ok: false, + results: [], + error: "boom", + }), + ).toBe("error"); + }); + + it("THE BUG: empty results with ok:true is an ERROR, not a success", () => { + // Old code silently succeeded here → stuck `processing` rows. + expect(classifyIndividualWorkerResult({ ok: true, results: [] })).toBe( + "error", + ); + expect( + classifyIndividualWorkerResult({ ok: true, results: [undefined] }), + ).toBe("error"); + expect(classifyIndividualWorkerResult({})).toBe("error"); + }); + + it("treats a result with no status as unexplainable (error)", () => { + expect( + classifyIndividualWorkerResult({ + ok: true, + results: [{ flags: [] }], + }), + ).toBe("error"); + }); +});