diff --git a/.hermes/plans/2026-08-22_nonissue-optimizations.md b/.hermes/plans/2026-08-22_nonissue-optimizations.md new file mode 100644 index 0000000..493d0a9 --- /dev/null +++ b/.hermes/plans/2026-08-22_nonissue-optimizations.md @@ -0,0 +1,33 @@ +# Optimisasi "non-issue" AI analysis pipeline + +## Scope +Dua item yang sebelumnya dinyatakan non-issue, kini dioptimalkan + 1 bug ordering +yang ditemukan saat menelusuri: + +1. **pickBatchWithinBudget: skip → break.** Pesan diurutkan `created_at ASC` + oleh DB. Setelah budget habis, pesan berikutnya pasti lebih besar/lebih kecil + arbitrer — skip-then-take menghasilkan batch non-kontigu (ada gap analisis + di tengah timeline). Ubah jadi stop at first overflow (break) supaya prefix + kronologis utuh; sisanya otomatis diambil gelombang berikutnya + (`shouldScheduleNext` sudah selalu true setelah sukses). +2. **max_tokens dinamis.** Hard-coded 16384 di llmCaller.ts → parameter + opsional `maxTokens?`; default tetap 16384. Caller text/media batch pass + nilai berbasis ukuran prompt (tiktoken) dengan floor/ceiling. +3. **Bug ordering UPDATE..RETURNING (bonus).** messagesAnalysis.ts + `getPendingMessagesByConversation`: SELECT ids di-order `created_at ASC` + tapi UPDATE...RETURNING tanpa ORDER BY → urutan rows balik tidak + terjamin. Konsumen pakai messages[0] sebagai anchor konteks + (beforeCreatedAt) dan pickBatchWithinBudget asumsi urutan. Fix: re-sort in + JS by created_at (stable) sebelum return. + +## Files touched +- src/modules/ai-moderation/batchProcessor.ts — break bukan skip; test baru. +- src/modules/ai-moderation/llmCaller.ts — param maxTokens. +- src/modules/ai-moderation/textBatchProcessor.ts / mediaBatchProcessor.ts — + hitung token prompt & pass maxTokens. +- src/modules/message-capture/messagesAnalysis.ts — sort hasil RETURNING. +- tests/batchBudget.test.ts — baru. + +## Verification +cd services/discord-gateway && bun run typecheck && bun run lint && bun run test +lalu commit+push, watch GHA, restart service via deploy pipeline. diff --git a/services/discord-gateway/src/modules/ai-moderation/batchBudget.ts b/services/discord-gateway/src/modules/ai-moderation/batchBudget.ts new file mode 100644 index 0000000..6f339e8 --- /dev/null +++ b/services/discord-gateway/src/modules/ai-moderation/batchBudget.ts @@ -0,0 +1,45 @@ +/** + * batchBudget.ts + * + * Pure batch-sizing helper extracted from batchProcessor.ts so it can be + * unit-tested without pulling in the Piscina worker pool, message store, + * or any other side-effectful import chain. + */ +import type { MessageRecord } from "../message-capture/types.js"; + +/** Token estimator contract (satisfied by conversationContext.estimateTokens). */ +export type TokenEstimator = (text: string) => number; + +/** + * Picks a batch of messages within a token budget. + * `tokensPerMessage` accounts for JSON structure overhead around each entry. + * The estimator is injected so this stays a pure function — callers in the + * batch pipeline pass the tiktoken-based estimateTokens. + */ +export function pickBatchWithinBudget( + messages: MessageRecord[], + maxTokens: number, + tokensPerMessage: number, + estimateTokens: TokenEstimator, +): MessageRecord[] { + const batch: MessageRecord[] = []; + let usedTokens = 0; + + for (const msg of messages) { + const content = msg.edited_content ?? msg.content; + const msgTokens = estimateTokens(content) + tokensPerMessage; + + // Stop at the first overflow instead of skipping: input is ordered + // created_at ASC, so a contiguous chronological prefix keeps the batch + // gap-free. Skipped-over messages would leave unanalyzed holes mid- + // timeline; anything past the budget is picked up by the next wave + // (processBatch always re-schedules after success). + if (usedTokens + msgTokens > maxTokens) { + break; + } + batch.push(msg); + usedTokens += msgTokens; + } + + return batch; +} diff --git a/services/discord-gateway/src/modules/ai-moderation/batchProcessor.ts b/services/discord-gateway/src/modules/ai-moderation/batchProcessor.ts index 3190b85..a435f0a 100644 --- a/services/discord-gateway/src/modules/ai-moderation/batchProcessor.ts +++ b/services/discord-gateway/src/modules/ai-moderation/batchProcessor.ts @@ -3,6 +3,7 @@ import { config } from "../../shared/config/config.js"; 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 { workerPool } from "./circuitBreaker.js"; import { estimateTokens } from "./conversationContext.js"; import { @@ -39,30 +40,21 @@ export let activeRequests = 0; /** * Picks a batch of messages within a token budget. - * `tokensPerMessage` accounts for JSON structure overhead around each entry. - * Uses a rough character-based token estimate (avoids async formatMessageForPrompt - * since this function runs in a synchronous promise chain). + * Thin wrapper over the pure helper in batchBudget.ts (kept here so the + * existing import surface stays stable); passes the tiktoken-based + * estimateTokens. See batchBudget.ts for the overflow-stopping semantics. */ export function pickBatchWithinBudget( messages: MessageRecord[], maxTokens: number, tokensPerMessage: number, ): MessageRecord[] { - const batch: MessageRecord[] = []; - let usedTokens = 0; - - for (const msg of messages) { - const content = msg.edited_content ?? msg.content; - // Accurate token count via tiktoken (+ overhead for JSON structure) - const msgTokens = estimateTokens(content) + tokensPerMessage; - - if (usedTokens + msgTokens <= maxTokens) { - batch.push(msg); - usedTokens += msgTokens; - } - } - - return batch; + return pickBatchWithinBudgetPure( + messages, + maxTokens, + tokensPerMessage, + estimateTokens, + ); } // --------------------------------------------------------------------------- diff --git a/services/discord-gateway/src/modules/ai-moderation/llmCaller.ts b/services/discord-gateway/src/modules/ai-moderation/llmCaller.ts index caed53f..a177db2 100644 --- a/services/discord-gateway/src/modules/ai-moderation/llmCaller.ts +++ b/services/discord-gateway/src/modules/ai-moderation/llmCaller.ts @@ -49,6 +49,10 @@ export async function callModerationLLM( targetIds: string[], label: string, signal?: AbortSignal, + // Output-side token cap. Defaults to the previous hard-coded value; batch + // callers pass a prompt-derived ceiling so small batches don't reserve a + // 16k completion budget (some routers pre-allocate KV cache per max_tokens). + maxTokens?: number, ): Promise<{ results: AnalysisResult[]; raw: ChatCompletion | null; @@ -75,7 +79,7 @@ export async function callModerationLLM( ]; const completion = await llmChat({ messages, - max_tokens: 16384, + max_tokens: maxTokens ?? 16384, jsonResponse: { type: "json_object" }, retries: 0, signal, diff --git a/services/discord-gateway/src/modules/ai-moderation/mediaBatchProcessor.ts b/services/discord-gateway/src/modules/ai-moderation/mediaBatchProcessor.ts index 843af45..ee3be02 100644 --- a/services/discord-gateway/src/modules/ai-moderation/mediaBatchProcessor.ts +++ b/services/discord-gateway/src/modules/ai-moderation/mediaBatchProcessor.ts @@ -13,6 +13,7 @@ import type { MessageRecord, } from "../message-capture/types.js"; import { getChannelCulture } from "./channelCultureStore.js"; +import { estimateTokens } from "./conversationContext.js"; import type { RetryState } from "./llmCaller.js"; import { callModerationLLM } from "./llmCaller.js"; import { prepareMediaMessage } from "./mediaAnalysisClient.js"; @@ -87,11 +88,22 @@ export async function runMediaBatch( timeoutId.unref(); try { + // Output budget scales with the prompt (see textBatchProcessor): small + // media batches don't need the full 16k completion window. + const promptEstimate = + 2000 + + estimateTokens(userContent) + + targets.reduce((sum, m) => sum + estimateTokens(m.content ?? "") + 50, 0); + const dynamicMaxTokens = Math.min( + 16384, + Math.max(2048, Math.ceil(promptEstimate * 1.5)), + ); const result = await callModerationLLM( async (_state: RetryState) => ({ system: systemText, user: userContent }), targetIds, `media-batch:${targetIds.length}msgs`, abortController.signal, + dynamicMaxTokens, ); log.info( { mediaCount: targets.length, resultCount: result.results.length }, diff --git a/services/discord-gateway/src/modules/ai-moderation/textBatchProcessor.ts b/services/discord-gateway/src/modules/ai-moderation/textBatchProcessor.ts index a582b5a..2904ee6 100644 --- a/services/discord-gateway/src/modules/ai-moderation/textBatchProcessor.ts +++ b/services/discord-gateway/src/modules/ai-moderation/textBatchProcessor.ts @@ -14,6 +14,7 @@ import type { MessageRecord, } from "../message-capture/types.js"; import { getChannelCulture } from "./channelCultureStore.js"; +import { estimateTokens } from "./conversationContext.js"; import type { ModerationPromptContent, RetryState } from "./llmCaller.js"; import { callModerationLLM } from "./llmCaller.js"; import { analyzeSingleMediaImage } from "./mediaAnalysisClient.js"; @@ -341,11 +342,29 @@ export async function runTextOnlyBatch( let batchResult: { results: AnalysisResult[]; raw: unknown }; try { + // Output budget scales with the prompt: the JSON verdict block is + // roughly proportional to message count, so a small sub-batch doesn't + // need to reserve a full 16k completion window. Estimated here from + // raw materials (system/rules baseline ~2k + context + message + // bodies) instead of inside buildContent, because max_tokens must be + // known at call time. + const subBatchPromptEstimate = + 2000 + + estimateTokens(contextBlock ?? "") + + batch.reduce( + (sum, m) => sum + estimateTokens(m.edited_content ?? m.content) + 50, + 0, + ); + const dynamicMaxTokens = Math.min( + 16384, + Math.max(2048, Math.ceil(subBatchPromptEstimate * 1.5)), + ); batchResult = await callModerationLLM( buildContent, targetIds, `text-batch-${i + 1}`, abortController.signal, + dynamicMaxTokens, ); } catch (err: any) { if (err.name === "AbortError" || abortController.signal.aborted) { diff --git a/services/discord-gateway/src/modules/message-capture/messagesAnalysis.ts b/services/discord-gateway/src/modules/message-capture/messagesAnalysis.ts index c2161c4..d9977ae 100644 --- a/services/discord-gateway/src/modules/message-capture/messagesAnalysis.ts +++ b/services/discord-gateway/src/modules/message-capture/messagesAnalysis.ts @@ -246,6 +246,14 @@ export class MessagesAnalysis { .returning(); }); + // UPDATE..RETURNING has no guaranteed row order (Postgres returns rows + // in physical update order). Consumers rely on chronological order: + // batchScheduler/pickBatchWithinBudget treat the array as a created_at + // ASC prefix, and the context anchor uses messages[0].created_at. + rows.sort( + (a, b) => + (a as MessageRecord).created_at - (b as MessageRecord).created_at, + ); return rows as MessageRecord[]; } catch (error) { this.logger.error( @@ -386,6 +394,13 @@ export class MessagesAnalysis { .returning(); }); + // Same UPDATE..RETURNING ordering guarantee as above: re-sort to + // created_at ASC so the individual fallback path also sees a + // chronological array. + rows.sort( + (a, b) => + (a as MessageRecord).created_at - (b as MessageRecord).created_at, + ); return rows as MessageRecord[]; } catch (error) { this.logger.error( diff --git a/services/discord-gateway/tests/batchBudget.test.ts b/services/discord-gateway/tests/batchBudget.test.ts new file mode 100644 index 0000000..8b4b5f6 --- /dev/null +++ b/services/discord-gateway/tests/batchBudget.test.ts @@ -0,0 +1,91 @@ +import { describe, expect, it } from "vitest"; +import { + pickBatchWithinBudget, + type TokenEstimator, +} from "../src/modules/ai-moderation/batchBudget.js"; +import type { MessageRecord } from "../src/modules/message-capture/types.js"; + +// Deterministic estimator: 1 token per character. Keeps the budget math +// exact regardless of tiktoken behavior (the real estimator is injected at +// the call site — see batchProcessor.ts). +const estimate: TokenEstimator = (text: string) => text.length; + +function msg(id: string, content: string, createdAt: number): MessageRecord { + return { + id, + guild_id: "g", + channel_id: "c", + thread_id: null, + user_id: "u", + username: "user", + avatar_url: null, + content, + edited_content: null, + created_at: createdAt, + edited_at: null, + deleted_at: null, + type: "text", + is_reply: false, + is_forward: false, + is_crosspost: false, + reference_message_id: null, + reference_channel_id: null, + reference_guild_id: null, + metadata: null, + }; +} + +describe("pickBatchWithinBudget", () => { + const TOKENS_PER_MESSAGE = 50; + + it("returns a contiguous chronological prefix — no gaps mid-timeline", () => { + // sizes: 100, 100, 400 (overflow), 10 + const messages = [ + msg("m1", "a".repeat(100), 1), + msg("m2", "b".repeat(100), 2), + msg("m3", "c".repeat(400), 3), + msg("m4", "d".repeat(10), 4), + ]; + const batch = pickBatchWithinBudget( + messages, + 500, + TOKENS_PER_MESSAGE, + estimate, + ); + + // m1(150)+m2(150)=300 fits; m3 would be 550 > 500 → stop. + // m4 must NOT be picked even though it alone fits (no timeline gap). + expect(batch.map((m) => m.id)).toEqual(["m1", "m2"]); + }); + + it("includes a message that exactly hits the budget", () => { + const messages = [msg("m1", "a".repeat(450), 1)]; + const batch = pickBatchWithinBudget( + messages, + 500, + TOKENS_PER_MESSAGE, + estimate, + ); + expect(batch.map((m) => m.id)).toEqual(["m1"]); + }); + + it("returns empty when the first message alone exceeds the budget", () => { + const messages = [ + msg("big", "x".repeat(1000), 1), + msg("m2", "y".repeat(10), 2), + ]; + const batch = pickBatchWithinBudget( + messages, + 500, + TOKENS_PER_MESSAGE, + estimate, + ); + expect(batch).toHaveLength(0); + }); + + it("handles empty input", () => { + expect( + pickBatchWithinBudget([], 500, TOKENS_PER_MESSAGE, estimate), + ).toEqual([]); + }); +});