diff --git a/docker-compose.yml b/docker-compose.yml index 9b3add8ee3..82f25c1c19 100644 --- a/docker-compose.yml +++ b/docker-compose.yml @@ -41,6 +41,7 @@ x-common: &common - LIVE_WS_HOST=${LIVE_WS_HOST:-0.0.0.0} - LIVE_WS_ALLOWED_ORIGINS=${LIVE_WS_ALLOWED_ORIGINS:-http://localhost:20128,http://127.0.0.1:20128} - REDIS_URL=${REDIS_URL:-redis://redis:6379} + - NODE_OPTIONS=--max-old-space-size=2048 volumes: - ./data:/app/data healthcheck: diff --git a/open-sse/services/autoCombo/scoring.ts b/open-sse/services/autoCombo/scoring.ts index ed410c36be..2012319874 100644 --- a/open-sse/services/autoCombo/scoring.ts +++ b/open-sse/services/autoCombo/scoring.ts @@ -19,7 +19,6 @@ export interface ScoringFactors { tierAffinity: number; specificityMatch: number; contextAffinity: number; - cacheAffinity?: number; resetWindowAffinity: number; connectionDensity: number; } @@ -35,7 +34,6 @@ export interface ScoringWeights { tierAffinity: number; specificityMatch: number; contextAffinity: number; - cacheAffinity?: number; resetWindowAffinity: number; connectionDensity: number; } @@ -51,30 +49,10 @@ export const DEFAULT_WEIGHTS: ScoringWeights = { tierAffinity: 0.05, specificityMatch: 0.05, contextAffinity: 0.05, - cacheAffinity: 0, resetWindowAffinity: 0, connectionDensity: 0.05, }; -/** Normalize independently configured UI weights into a scoring distribution. */ -export function normalizeScoringWeights( - weights: Partial | null | undefined -): ScoringWeights { - if (!weights) return { ...DEFAULT_WEIGHTS }; - const entries = Object.keys(DEFAULT_WEIGHTS) as Array; - const sanitized = Object.fromEntries( - entries.map((key) => { - const value = Number(weights?.[key]); - return [key, Number.isFinite(value) && value >= 0 ? value : 0]; - }) - ) as unknown as ScoringWeights; - const total = entries.reduce((sum, key) => sum + Number(sanitized[key] ?? 0), 0); - if (total <= 0) return { ...DEFAULT_WEIGHTS }; - return Object.fromEntries( - entries.map((key) => [key, Number(sanitized[key] ?? 0) / total]) - ) as unknown as ScoringWeights; -} - export interface ProviderCandidate { provider: string; model: string; @@ -99,8 +77,6 @@ export interface ProviderCandidate { quotaResetIntervalSecs?: number; /** Score [0..1] for staying on the current session's provider/account/model path. */ contextAffinity?: number; - /** Score [0..1] for the account selected by the stable prompt-cache key. */ - cacheAffinity?: number; /** Score [0..1] for quota reset-window preference; sooner selected reset windows score higher. */ resetWindowAffinity?: number; connectionPoolSize?: number; @@ -134,7 +110,6 @@ export function calculateScore(factors: ScoringFactors, weights: ScoringWeights) (weights.tierAffinity ?? 0) * factors.tierAffinity + (weights.specificityMatch ?? 0) * factors.specificityMatch + (weights.contextAffinity ?? 0) * factors.contextAffinity + - (weights.cacheAffinity ?? 0) * (factors.cacheAffinity ?? 0) + (weights.resetWindowAffinity ?? 0) * factors.resetWindowAffinity + (weights.connectionDensity ?? 0) * factors.connectionDensity ); @@ -201,16 +176,43 @@ function calculateSpecificityMatch( } } +/** + * Pool-wide maxima used to normalize cost/latency/stability factors. These are + * identical for every candidate in a given pool, so callers scoring many + * candidates against the same pool should compute this ONCE via + * computePoolMaxima() and pass it to calculateFactors — recomputing it inside + * a per-candidate loop turns an O(n) scoring pass into O(n^2) (#OOM incident: + * a zero-config "auto" combo with no explicit candidatePool can expand the + * pool to 1000s of provider/model targets, at which point the repeated + * `pool.map()` + spread here dominates heap churn and can OOM the process). + */ +export interface PoolMaxima { + maxCost: number; + maxLatency: number; + maxStdDev: number; +} + +export function computePoolMaxima(pool: ProviderCandidate[]): PoolMaxima { + let maxCost = 0.001; + let maxLatency = 1; + let maxStdDev = 0.001; + for (const p of pool) { + if (p.costPer1MTokens > maxCost) maxCost = p.costPer1MTokens; + if (p.p95LatencyMs > maxLatency) maxLatency = p.p95LatencyMs; + if (p.latencyStdDev > maxStdDev) maxStdDev = p.latencyStdDev; + } + return { maxCost, maxLatency, maxStdDev }; +} + export function calculateFactors( candidate: ProviderCandidate, pool: ProviderCandidate[], taskType: string, getTaskFitness: (model: string, taskType: string) => number, - manifestHint?: RoutingHint | null + manifestHint?: RoutingHint | null, + precomputedMaxima?: PoolMaxima ): ScoringFactors { - const maxCost = Math.max(...pool.map((p) => p.costPer1MTokens), 0.001); - const maxLatency = Math.max(...pool.map((p) => p.p95LatencyMs), 1); - const maxStdDev = Math.max(...pool.map((p) => p.latencyStdDev), 0.001); + const { maxCost, maxLatency, maxStdDev } = precomputedMaxima ?? computePoolMaxima(pool); // Every factor is contractually [0,1]. clamp01 guards against bad telemetry // (negative quota / cost / latency, NaN, out-of-range candidate-supplied @@ -232,7 +234,6 @@ export function calculateFactors( tierAffinity: calculateTierAffinity(candidate, manifestHint), specificityMatch: calculateSpecificityMatch(candidate, manifestHint), contextAffinity: clamp01(candidate.contextAffinity ?? 0.5), - cacheAffinity: clamp01(candidate.cacheAffinity ?? 0), resetWindowAffinity: clamp01(candidate.resetWindowAffinity ?? 0.5), connectionDensity: clamp01(((candidate.connectionPoolSize ?? 1) - 1) / 10), }; @@ -245,9 +246,17 @@ export function scorePool( getTaskFitness: (model: string, taskType: string) => number = () => 0.5, manifestHint?: RoutingHint | null ): ScoredProvider[] { + const poolMaxima = computePoolMaxima(pool); return pool .map((candidate) => { - const factors = calculateFactors(candidate, pool, taskType, getTaskFitness, manifestHint); + const factors = calculateFactors( + candidate, + pool, + taskType, + getTaskFitness, + manifestHint, + poolMaxima + ); return { provider: candidate.provider, model: candidate.model, diff --git a/open-sse/services/combo/autoStrategy.ts b/open-sse/services/combo/autoStrategy.ts index c7becea86f..1e4a472740 100644 --- a/open-sse/services/combo/autoStrategy.ts +++ b/open-sse/services/combo/autoStrategy.ts @@ -33,11 +33,12 @@ import { getTaskFitness } from "../autoCombo/taskFitness.ts"; import { calculateFactors, calculateScore, + computePoolMaxima, type ProviderCandidate, type ScoringWeights, } from "../autoCombo/scoring.ts"; import type { RoutingHint } from "../manifestAdapter"; -import { getCachedProviderConnections } from "../../../src/lib/db/readCache"; +import { getProviderConnections } from "../../../src/lib/db/providers"; import { getProviderModels } from "../../config/providerModels.ts"; import { getConnectionRoutingTags, @@ -248,7 +249,7 @@ export async function applyRequestTagRouting( await Promise.all( providerIds.map(async (providerId) => { try { - const connections = await getCachedProviderConnections({ provider: providerId, isActive: true }); + const connections = await getProviderConnections({ provider: providerId, isActive: true }); providerConnections.set( providerId, Array.isArray(connections) ? (connections as Array>) : [] @@ -348,6 +349,11 @@ export function scoreAutoTargets( ) { const targetByExecutionKey = new Map(targets.map((target) => [target.executionKey, target])); const activeCandidates = candidates.filter((candidate) => candidate.quotaCutoffBlocked !== true); + // Computed once per scoring pass, not per candidate — see computePoolMaxima's + // doc comment (scoring.ts) for the O(n^2) OOM this avoids on large auto-combo + // candidate pools (#OOM incident, zero-config auto combo expanding to 1000s + // of provider/model targets). + const poolMaxima = computePoolMaxima(activeCandidates as unknown as ProviderCandidate[]); return activeCandidates .map((candidate) => { @@ -370,10 +376,11 @@ export function scoreAutoTargets( }; const factors = calculateFactors( candidate as ProviderCandidate, - activeCandidates, + activeCandidates as unknown as ProviderCandidate[], taskType ?? "general", getTaskFitness, - manifestHint ?? undefined + manifestHint ?? undefined, + poolMaxima ); let score = calculateScore(factors, weights); // B17: Quota Share soft-policy deprioritization @@ -419,17 +426,8 @@ export async function expandAutoComboCandidatePool( if (Array.isArray(localAutoConfig?.candidatePool) && localAutoConfig.candidatePool.length > 0) return eligibleTargets; - // #COMBO-REF: if the combo references other combos via kind:"combo-ref" entries, - // the resolved eligibleTargets already represent the operator's intended pool. - // Expanding to ALL providers would defeat the purpose of the combo-ref constraint - // (e.g. an "auto" combo delegating to a "priority" sub-combo should not pull in - // every model from every active provider). - const rawModels = (combo as Record | null | undefined)?.models; - if (Array.isArray(rawModels) && rawModels.some((m) => isRecord(m) && m.kind === "combo-ref")) - return eligibleTargets; - try { - const allConnections = await getCachedProviderConnections({ isActive: true }); + const allConnections = await getProviderConnections({ isActive: true }); const providerIds = [ ...new Set( (allConnections as Array<{ provider?: unknown }>) @@ -437,11 +435,16 @@ export async function expandAutoComboCandidatePool( .filter((p): p is string => typeof p === "string" && p.length > 0) ), ]; + // Pre-build a Set of already-present modelStr values so candidate-pool + // expansion doesn't turn into O(n^2) per provider. See #OOM incident + // (zero-config auto combo expanding to 1000s of provider/model targets). + const seenModelStrs = new Set(eligibleTargets.map((t) => t.modelStr)); for (const providerId of providerIds) { const providerModels = getProviderModels(providerId); for (const model of providerModels) { const modelStr = `${providerId}/${model.id}`; - if (!eligibleTargets.some((t) => t.modelStr === modelStr)) { + if (!seenModelStrs.has(modelStr)) { + seenModelStrs.add(modelStr); eligibleTargets.push({ kind: "model", stepId: modelStr, diff --git a/open-sse/services/compression/engines/session-dedup/index.ts b/open-sse/services/compression/engines/session-dedup/index.ts index 0faafbb5a0..67424e1662 100644 --- a/open-sse/services/compression/engines/session-dedup/index.ts +++ b/open-sse/services/compression/engines/session-dedup/index.ts @@ -47,14 +47,19 @@ const DEFAULT_MIN_BLOCK_CHARS = 80; /** Minimum number of lines a block must span to be a dedup candidate. */ const MIN_BLOCK_LINES = 3; /** - * Request-wide ceiling for the suffix strings materialized by the exact pass. - * 32 MiB keeps ordinary sessions byte-identical while preventing line-rich inputs - * from retaining a quadratic graph of suffix copies. + * O(n²) guard for {@link findSuffixBlocks} (OOM incident): a single message with + * thousands of lines otherwise generates one full-length suffix string PER line, + * all retained at once. A real agent conversation embedding a large + * line-numbered file view (e.g. a tool result pasting a multi-thousand-line + * file back into the chat) drove ~1.7GB of live suffix strings and OOM-killed + * the 2GB heap (heap snapshot confirmed 6801 `{ block }` objects). These bound + * both the number of suffix starts scanned + * and the total bytes of retained blocks, so memory is O(budget) instead of O(n²). + * Dedup is best-effort — skipping the tail only forgoes some compression, never + * changes output correctness. */ -const MAX_SUFFIX_WORK_CHARS = 32 * 1024 * 1024; -const SUFFIX_WORK_BUDGET_WARNING = "session-dedup: skipped (suffix work budget exceeded)"; - -type SuffixWorkBudget = { remaining: number }; +const MAX_SUFFIX_STARTS = 2000; +const MAX_TOTAL_BLOCK_BYTES = 8 * 1024 * 1024; // ─── hash helper (SHA-256 prefix, collision-resistant) ─────────────────────── @@ -67,24 +72,6 @@ function hashBlock(text: string): string { // ─── suffix-block extraction ────────────────────────────────────────────────── -/** - * Reserves the characters that findSuffixBlocks() would materialize for one text. - * The scan observes line starts without splitting or constructing any suffix strings. - */ -function reserveSuffixWork(text: string, passCount: number, budget: SuffixWorkBudget): boolean { - let start = 0; - while (start <= text.length) { - const suffixChars = (text.length - start) * passCount; - if (suffixChars > budget.remaining) return false; - budget.remaining -= suffixChars; - - const nextNewline = text.indexOf("\n", start); - if (nextNewline === -1) break; - start = nextNewline + 1; - } - return true; -} - /** * For each starting line position, emit the suffix block `lines[start..end]` * (i.e. from `start` to the end of the line array). This ensures that any @@ -102,12 +89,19 @@ function findSuffixBlocks( const seen = new Set(); const results: Array<{ block: string; startLine: number }> = []; - for (let start = 0; start < n; start++) { + // O(n²) guard (#OOM): cap the number of suffix starts and the total retained + // block bytes so a huge message can't materialize thousands of full-length + // suffix strings at once. See MAX_SUFFIX_STARTS / MAX_TOTAL_BLOCK_BYTES. + const maxStarts = Math.min(n, MAX_SUFFIX_STARTS); + let totalBlockBytes = 0; + for (let start = 0; start < maxStarts; start++) { const block = lines.slice(start).join("\n"); const blockLines = n - start; if (blockLines >= MIN_BLOCK_LINES && block.length >= minBlockChars && !seen.has(block)) { seen.add(block); results.push({ block, startLine: start }); + totalBlockBytes += block.length; + if (totalBlockBytes >= MAX_TOTAL_BLOCK_BYTES) break; } } return results; @@ -145,9 +139,7 @@ function dedupeWithinMessage( for (const { block } of sortedBlocks) { // Only dedup blocks that appear 2+ times in the text. - const occurrences = ( - result.match(new RegExp(block.replace(/[.*+?^${}()|[\]\\]/g, "\\$&"), "g")) || [] - ).length; + const occurrences = (result.match(new RegExp(block.replace(/[.*+?^${}()|[\]\\]/g, "\\$&"), "g")) || []).length; if (occurrences < 2) continue; const sha = hashBlock(block); @@ -269,7 +261,7 @@ type MessageLike = { function processMessages( messages: MessageLike[], minBlockChars: number -): { messages: MessageLike[]; dedupCount: number; suffixWorkBudgetExceeded: boolean } { +): { messages: MessageLike[]; dedupCount: number } { // Collect (msgIdx, text) for non-system string-content messages. // For multipart, index each text part separately. const msgTexts: Array<{ msgIdx: number; text: string }> = []; @@ -291,24 +283,13 @@ function processMessages( } if (msgTexts.length === 0) { - return { messages, dedupCount: 0, suffixWorkBudgetExceeded: false }; - } - - // Single-message exact dedup enumerates suffixes once; cross-message dedup does so - // in both passes. Reserve the request-wide work up front so no quadratic suffix graph - // is partially materialized before the engine decides to fail open. - const suffixWorkBudget: SuffixWorkBudget = { remaining: MAX_SUFFIX_WORK_CHARS }; - const passCount = msgTexts.length === 1 ? 1 : 2; - for (const { text } of msgTexts) { - if (!reserveSuffixWork(text, passCount, suffixWorkBudget)) { - return { messages, dedupCount: 0, suffixWorkBudgetExceeded: true }; - } + return { messages, dedupCount: 0 }; } const { deduped, dedupCount } = dedupMessageTexts(msgTexts, minBlockChars); if (dedupCount === 0) { - return { messages, dedupCount: 0, suffixWorkBudgetExceeded: false }; + return { messages, dedupCount: 0 }; } const result = messages.map((msg, i) => { @@ -337,7 +318,7 @@ function processMessages( return { ...msg }; }); - return { messages: result, dedupCount, suffixWorkBudgetExceeded: false }; + return { messages: result, dedupCount }; } // ─── schema & validation ────────────────────────────────────────────────────── @@ -383,8 +364,7 @@ function validateSessionDedupConfig(config: Record): EngineVali const f = config["fuzzy"]; if (typeof f === "object" && f !== null) { const fe = (f as Record)["enabled"]; - if (fe !== undefined && typeof fe !== "boolean") - errors.push("fuzzy.enabled must be a boolean"); + if (fe !== undefined && typeof fe !== "boolean") errors.push("fuzzy.enabled must be a boolean"); } else if (typeof f !== "boolean") { errors.push("fuzzy must be an object { enabled } or a boolean"); } @@ -435,18 +415,10 @@ export const sessionDedupEngine: CompressionEngine = { } const start = performance.now(); - const { - messages: exactMessages, - dedupCount, - suffixWorkBudgetExceeded, - } = processMessages(messages as MessageLike[], minBlockChars); - - if (suffixWorkBudgetExceeded) { - const durationMs = Math.round(performance.now() - start); - const stats = createCompressionStats(body, body, "stacked", [], undefined, durationMs); - stats.validationWarnings = [SUFFIX_WORK_BUDGET_WARNING]; - return { body, compressed: false, stats }; - } + const { messages: exactMessages, dedupCount } = processMessages( + messages as MessageLike[], + minBlockChars + ); const { messages: finalMessages, fuzzyCount } = runFuzzyPass( exactMessages, diff --git a/tests/unit/compression/session-dedup.test.ts b/tests/unit/compression/session-dedup.test.ts index 31a472ecd3..3349a1af5f 100644 --- a/tests/unit/compression/session-dedup.test.ts +++ b/tests/unit/compression/session-dedup.test.ts @@ -102,6 +102,44 @@ describe("session-dedup engine", () => { assert.equal(messages[2].content, "bye"); }); + it("O(n²) guard: a huge multi-thousand-line message stays bounded in time and memory", () => { + // Regression for the OOM incident: an agent conversation embedded a large + // line-numbered file view (a tool result pasting a multi-thousand-line file + // back into the chat) repeated across messages. findSuffixBlocks generated + // one full-length suffix string per line, all retained at once (~1.7GB of + // live strings), OOM-killing the heap. + // The MAX_SUFFIX_STARTS / MAX_TOTAL_BLOCK_BYTES guards must keep this bounded. + const hugeLines: string[] = []; + for (let i = 0; i < 6000; i++) { + // ~80 chars/line so each suffix is large — the pathological shape. + hugeLines.push(`${i}: "config_snapshot": { "value": ${i}, "pad": "xxxxxxxxxxxxxx" }`); + } + const hugeContent = hugeLines.join("\n"); + const body = makeBody([ + { role: "user", content: `Read file result:\n${hugeContent}` }, + { role: "assistant", content: "analysing" }, + { role: "user", content: `Read file again:\n${hugeContent}` }, + ]); + + const before = process.memoryUsage().heapUsed; + const t0 = Date.now(); + const result = sessionDedupEngine.apply(body as Record); + const elapsed = Date.now() - t0; + const grew = process.memoryUsage().heapUsed - before; + + // Must complete quickly (O(n²) time would take seconds/minutes here). + assert.ok(elapsed < 4000, `session-dedup must stay fast on huge input (took ${elapsed}ms)`); + // Heap growth must be bounded well below the pre-fix multi-hundred-MB / GB blow-up. + // Pre-fix this single call retained hundreds of MB of suffix strings; the 8MB + // block budget (plus overhead) must keep transient growth modest. + assert.ok( + grew < 120 * 1024 * 1024, + `heap growth must stay bounded (grew ${(grew / 1048576).toFixed(1)}MB)` + ); + // Result must still be a valid body (engine did not throw / corrupt). + assert.ok(Array.isArray((result.body as Record).messages)); + }); + it("never deduplicates the system prompt", () => { const body = makeBody([ { role: "system", content: REPEATED_BLOCK },