fix(compression): bound session-dedup suffix-block scan to prevent OOM

findSuffixBlocks() generated one full-length suffix string per line of a message, retaining all of them simultaneously - O(n^2) memory with no cap. An agent conversation embedding a large line-numbered file view (a tool result pasting a multi-thousand-line file back into the chat) drove ~1.7GB of live strings, exhausting the 2GB heap and crash-looping the container. Bound the scan to MAX_SUFFIX_STARTS (2000) suffix starts and MAX_TOTAL_BLOCK_BYTES (8MB) of retained block content. Dedup is best-effort - skipping the tail only forgoes some compression, never changes output correctness. Confirmed via heap snapshot analysis on the actual OOM: 93% of the 2GB heap was `{ block }` objects from this function. Also extracts computePoolMaxima() out of the per-candidate scoring loop in autoCombo/scoring.ts and combo/autoStrategy.ts, removing a redundant O(n^2) in auto-combo candidate scoring found during the same investigation (not the OOM's cause, but genuine inefficiency). docker-compose.yml: NODE_OPTIONS is now set explicitly in `environment:`, since OMNIROUTE_MEMORY_MB from .env never actually reaches NODE_OPTIONS at runtime (the Dockerfile bakes it at build time).
2026-07-26 09:52:11 +03:00 · 2026-07-24 11:59:11 -03:00
parent 9994e00763
commit f4b3b0c5c6
5 changed files with 127 additions and 104 deletions
--- 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:
--- 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<ScoringWeights> | null | undefined
-): ScoringWeights {
-  if (!weights) return { ...DEFAULT_WEIGHTS };
-  const entries = Object.keys(DEFAULT_WEIGHTS) as Array<keyof ScoringWeights>;
-  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,
--- 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<Record<string, unknown>>) : []
@@ -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<string, unknown> | 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,
--- 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<string>();
  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<string, unknown>): EngineVali
    const f = config["fuzzy"];
    if (typeof f === "object" && f !== null) {
      const fe = (f as Record<string, unknown>)["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,
--- 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<string, unknown>);
+    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<string, unknown>).messages));
+  });
+
  it("never deduplicates the system prompt", () => {
    const body = makeBody([
      { role: "system", content: REPEATED_BLOCK },