mirror of
https://github.com/diegosouzapw/OmniRoute.git
synced 2026-09-17 04:12:17 +03:00
perf(compression): OOM mitigations for large payload hashing, memoization, and token estimation (#7847) (#11844)
* perf(compression): memory and OOM mitigations for large payload hashing and token estimation * fix(compression): implement getMemoStats observability for result memo (#7847) Adds the missing memo observability layer referenced by tests/unit/compression/oom-memo-memory.test.ts and the monitoring API: - resultMemo.ts: lifetime hit/miss counters + bounded time-ordered ring buffer (10k entries, ~90KB) powering 1m/5m/15m/1h hit-rate windows; getMemoStats() reports size/capacity/hits/misses/hitRate + windows. - memoLookup() tags served results with stats.memoHit = true. - clearMemoStore() also resets counters and the ring. - compression/index.ts re-exports getMemoStats for the monitoring route. - types.ts: optional memoHit field on CompressionStats. - New GET /api/monitoring/compression route exposing the stats snapshot (lightweight, no DB) for operators to track cache-hit efficiency. * fix(compression): align memo contract with upstream #11727 — return caller object, reset lookup counter in clearMemoStore * fix(compression): restore unwrapEventEnvelope in stream payload collector summaries The OOM-mitigation commit accidentally replaced unwrapEventEnvelope(evt.data) with asRecord(evt.data) in the summary builders and live push, breaking translate-mode {event, data} envelope unwrapping (clientPayload type detection) and failing 2 stream-payload-collector tests. Restored upstream semantics; kept the jsonLength OOM optimization as the only delta in this file. * refactor(compression): break down writeValue and writeEncodedString to pass complexity ratchets Refactors jsonSha256 internal helpers (writeValue, writeEncodedString) into small, single-responsibility sub-functions under the complexity threshold (max cyclomatic 15, max cognitive 15). Preserves exact JSON.stringify parity, circular reference guards on both arrays and plain objects, and escape behavior (all 530 relevant tests pass). * test(compression): make oom-memo heap assertion robust without expose-gc The CI unit-test shard runner does not pass --expose-gc, so global.gc is undefined and heapUsed can still momentarily hold GC-pending transients (observed 53.4 MiB after a 3MiB body). Gate the retained-heap assertion on forced collection being available (3 forced cycles for array buffers) instead of skipping it silently, and keep it fully active when --expose-gc is present. * fix(compression): restore worker-pool offload path in runCompressionAsync The OOM-mitigation refactor dropped the isCompressionWorkerEligible / runCompressionInWorker dispatch at the top of runCompressionAsync, silently removing the base's worker-thread offload for eligible large payloads. Restore the block exactly as on release/v3.8.51, ahead of the result-memo path, keeping the memoization and hashing improvements intact. * docs(api): document GET /api/monitoring/compression and log route errors via pino Add the new monitoring endpoint to docs/openapi.yaml following the neighboring System entries, and replace the route's console.error with the repo-standard pino logger. * fix(skills): regenerate omni-resilience and add changelog fragment Co-authored-by: diegosouzapw <8016841+diegosouzapw@users.noreply.github.com> --------- Co-authored-by: diegosouzapw <8016841+diegosouzapw@users.noreply.github.com> Co-authored-by: Andrian Balanescu <AndrianBalanescu@users.noreply.github.com> Co-authored-by: Diego Rodrigues de Sa e Souza <diegosouza.pw@gmail.com>
This commit is contained in:
Andrew B.
1
changelog.d/fixes/11844-compression-oom-mitigations.md
Normal file
1
changelog.d/fixes/11844-compression-oom-mitigations.md
Normal file
@@ -0,0 +1 @@
|
||||
- **perf(compression):** OOM mitigations for large payload hashing, memoization, and token estimation ([#11844](https://github.com/diegosouzapw/OmniRoute/pull/11844) — thanks @AndrianBalanescu)
|
||||
@@ -6199,6 +6199,22 @@ paths:
|
||||
"200":
|
||||
description: Health status
|
||||
|
||||
/api/monitoring/compression:
|
||||
get:
|
||||
tags: [System]
|
||||
summary: Get compression result-memo statistics
|
||||
description: >-
|
||||
In-process compression result-memo observability snapshot — size, capacity,
|
||||
lifetime hits/misses/hitRate plus 1m/5m/15m/1h windowed rates. Lightweight
|
||||
(no DB, no provider reads) companion to `GET /api/monitoring/health` intended
|
||||
for frequent polling. Sent with `Cache-Control: no-store, no-cache,
|
||||
must-revalidate`. Counters reset on process restart.
|
||||
responses:
|
||||
"200":
|
||||
description: Compression memo stats (`compression.memo` + `timestamp`)
|
||||
"503":
|
||||
description: Compression stats unavailable
|
||||
|
||||
/api/rate-limits:
|
||||
get:
|
||||
tags: [System]
|
||||
|
||||
@@ -11,7 +11,11 @@ import type {
|
||||
EngineValidationResult,
|
||||
} from "../types.ts";
|
||||
import { CODEX_RESPONSE_ITEM_META } from "../../bodyAdapter.ts";
|
||||
import { countTextTokens } from "../../../../../src/shared/utils/tiktokenCounter.ts";
|
||||
import {
|
||||
countTextTokens,
|
||||
MAX_EXACT_TOKEN_COUNT_CHARS,
|
||||
} from "../../../../../src/shared/utils/tiktokenCounter.ts";
|
||||
import { jsonLength, jsonLengthStrippingBase64DataUris } from "../../../../utils/jsonSize.ts";
|
||||
|
||||
const ENGINE_ID = "codex-responses";
|
||||
|
||||
@@ -19,6 +23,23 @@ function countCodexTokens(text: string): number {
|
||||
if (!text) return 0;
|
||||
return countTextTokens(text, { provider: "codex" });
|
||||
}
|
||||
|
||||
/** Codex-context token count for a whole body, skipping JSON.stringify on oversized
|
||||
* bodies: countTextTokens falls back to a char heuristic above MAX_EXACT_TOKEN_COUNT_CHARS,
|
||||
* so materializing a multi-MB string for the count is a pure OOM-class transient (#7847). */
|
||||
function countCodexTokensForBody(body: unknown): number {
|
||||
if (body === null || body === undefined) return 0;
|
||||
if (typeof body === "string") return countCodexTokens(body);
|
||||
if (jsonLength(body) > MAX_EXACT_TOKEN_COUNT_CHARS) {
|
||||
// Oversized bodies skip countTextTokens (which falls back to a char heuristic above
|
||||
// MAX_EXACT_TOKEN_COUNT_CHARS) to avoid materializing a multi-MB string (#7847). But the
|
||||
// exact path it replaces also stripped base64 data URIs first; the heuristic must too,
|
||||
// otherwise embedded screenshots inflate the reported token count and distort
|
||||
// savingsPercent. (The compression DECISION is unaffected either way.)
|
||||
return Math.ceil(jsonLengthStrippingBase64DataUris(body) / 4);
|
||||
}
|
||||
return countCodexTokens(JSON.stringify(body));
|
||||
}
|
||||
const SUPPORTED_TYPES = new Set([
|
||||
"function_call_output",
|
||||
"local_shell_call_output",
|
||||
@@ -274,8 +295,8 @@ export const codexResponsesEngine: CompressionEngine = {
|
||||
if (!changed) return { body, compressed: false, stats: null };
|
||||
const nextBody = { ...body, messages };
|
||||
const stats = createCompressionStats(body, nextBody, "codex-responses", [ENGINE_ID]);
|
||||
const originalTokens = countCodexTokens(JSON.stringify(body));
|
||||
const compressedTokens = countCodexTokens(JSON.stringify(nextBody));
|
||||
const originalTokens = countCodexTokensForBody(body);
|
||||
const compressedTokens = countCodexTokensForBody(nextBody);
|
||||
stats.originalTokens = originalTokens;
|
||||
stats.compressedTokens = compressedTokens;
|
||||
stats.savingsPercent =
|
||||
|
||||
@@ -162,17 +162,18 @@ export function applyHardBudget(
|
||||
// Distribute the aggregate budget proportionally per message so the SUM stays
|
||||
// ≤ target (passing the full target to each message would let an N-message body
|
||||
// come back N× over budget).
|
||||
let changed = false;
|
||||
const newMessages = messages.map((m) => {
|
||||
if (typeof m.content !== "string") return m;
|
||||
const msgTokens = countTextTokens(m.content, tokenizerContext);
|
||||
const perMsgTarget =
|
||||
totalTokens > 0 ? Math.floor(effectiveTarget * (msgTokens / totalTokens)) : effectiveTarget;
|
||||
const out = compressText(m.content, perMsgTarget, tokenizerContext);
|
||||
return out === m.content ? m : { ...m, content: out };
|
||||
if (out === m.content) return m;
|
||||
changed = true;
|
||||
return { ...m, content: out };
|
||||
});
|
||||
|
||||
const changed = newMessages.some((m, i) => JSON.stringify(m) !== JSON.stringify(messages[i]));
|
||||
|
||||
// Measure the result to detect when preserve-guarded content makes the target
|
||||
// unreachable, so callers are not silently left over budget.
|
||||
const usedMessages = changed ? newMessages : messages;
|
||||
|
||||
@@ -90,6 +90,8 @@ export {
|
||||
applyStackedCompressionAsync,
|
||||
} from "./strategySelector.ts";
|
||||
|
||||
export { getMemoStats, clearMemoStore, makeMemoKey, isDeterministicMode } from "./resultMemo.ts";
|
||||
|
||||
export type {
|
||||
CompressionEngine,
|
||||
CompressionEngineApplyOptions,
|
||||
|
||||
@@ -1,7 +1,6 @@
|
||||
import { createHash } from "node:crypto";
|
||||
|
||||
import { estimateCompressionTokens } from "./stats.ts";
|
||||
import type { CompressionResult, CompressionStats } from "./types.ts";
|
||||
import { jsonSha256 } from "../../utils/jsonHash.ts";
|
||||
|
||||
export interface LiveZoneOptions {
|
||||
principalId?: string;
|
||||
@@ -57,8 +56,15 @@ function serialize(value: unknown): string | null {
|
||||
}
|
||||
|
||||
function digest(value: unknown): string | null {
|
||||
const serialized = serialize(value);
|
||||
return serialized === null ? null : createHash("sha256").update(serialized).digest("hex");
|
||||
// jsonSha256 computes sha256hex(JSON.stringify(value)) WITHOUT materializing the
|
||||
// multi-MB string, avoiding the #7847 OOM-class transient on large tool-message
|
||||
// items (e.g. base64 screenshots). Throws on non-serializable values, matching
|
||||
// the previous JSON.stringify behavior which the caller treats as a miss.
|
||||
try {
|
||||
return jsonSha256(value);
|
||||
} catch {
|
||||
return null;
|
||||
}
|
||||
}
|
||||
|
||||
function cloneItems(items: unknown[]): unknown[] | null {
|
||||
|
||||
@@ -1,10 +1,52 @@
|
||||
import crypto from "node:crypto";
|
||||
import type { CompressionConfig, CompressionMode, CompressionResult } from "./types.ts";
|
||||
import { jsonSha256 } from "../../utils/jsonHash.ts";
|
||||
|
||||
export const MEMO_CAP = 5_000;
|
||||
|
||||
const memoMap = new Map<string, CompressionResult>();
|
||||
let lookupCountForTests = 0;
|
||||
let memoHits = 0;
|
||||
let memoMisses = 0;
|
||||
|
||||
// ── Windowed hit/miss ring buffer for time-bucketed stats ──────────────
|
||||
// Records each lookup outcome with a ms timestamp. getMemoStats scans the
|
||||
// ring to compute 1m/5m/15m/1h windows (like load average) so operators see
|
||||
// the *current* hit rate during a traffic spike, not a diluted all-time
|
||||
// average. Bounded memory: RING_CAP * ~9 bytes ≈ 90 KB, fixed-size array.
|
||||
const RING_CAP = 10_000;
|
||||
const ring: Array<{ ts: number; hit: boolean } | undefined> = new Array(RING_CAP);
|
||||
let ringHead = 0; // index of the NEXT write slot (wraps)
|
||||
let ringCount = 0; // entries written so far (clamped to RING_CAP)
|
||||
|
||||
function recordLookup(hit: boolean): void {
|
||||
ring[ringHead] = { ts: Date.now(), hit };
|
||||
ringHead = (ringHead + 1) % RING_CAP;
|
||||
if (ringCount < RING_CAP) ringCount++;
|
||||
}
|
||||
|
||||
/** Compute hits/misses/hitRate for lookups within the last `windowMs`. */
|
||||
function windowStats(windowMs: number): { hits: number; misses: number; hitRate: number } {
|
||||
const cutoff = Date.now() - windowMs;
|
||||
let hits = 0;
|
||||
let misses = 0;
|
||||
// Walk newest→oldest. The ring is time-ordered (oldest at head), so once
|
||||
// an entry is older than the cutoff every earlier one is too — early break.
|
||||
for (let k = 0; k < ringCount; k++) {
|
||||
const idx = (ringHead - 1 - k + RING_CAP) % RING_CAP;
|
||||
const e = ring[idx];
|
||||
if (!e) break;
|
||||
if (e.ts < cutoff) break;
|
||||
if (e.hit) hits++;
|
||||
else misses++;
|
||||
}
|
||||
const total = hits + misses;
|
||||
return {
|
||||
hits,
|
||||
misses,
|
||||
hitRate: total > 0 ? Math.round((hits / total) * 10000) / 100 : 0,
|
||||
};
|
||||
}
|
||||
|
||||
// Opt-IN whitelist (NOT opt-out): cache only engines proven pure + STATELESS across
|
||||
// requests. Excluded on purpose: `ccr` and `session-dedup` write to the cross-request
|
||||
@@ -41,7 +83,9 @@ export function makeMemoKey(
|
||||
model?: string,
|
||||
supportsVision?: boolean | null
|
||||
): string {
|
||||
const bodyHash = sha256hex(JSON.stringify(body));
|
||||
// Uses streaming jsonSha256 instead of sha256hex(JSON.stringify(body))
|
||||
// to avoid allocating multi-MB string transients on large agent payloads (#7847).
|
||||
const bodyHash = jsonSha256(body);
|
||||
|
||||
// #8137: Only include model + supportsVision in the cache key when the compression
|
||||
// result actually depends on them. The `lite` engine strips data:image URLs only when
|
||||
@@ -97,22 +141,74 @@ function boundedSet(key: string, value: CompressionResult): void {
|
||||
export function memoLookup(key: string): CompressionResult | null {
|
||||
lookupCountForTests++;
|
||||
const hit = memoMap.get(key);
|
||||
if (!hit) return null;
|
||||
if (!hit) {
|
||||
memoMisses++;
|
||||
recordLookup(false);
|
||||
return null;
|
||||
}
|
||||
memoHits++;
|
||||
recordLookup(true);
|
||||
// Return a clone so downstream mutation cannot corrupt the cached value.
|
||||
return JSON.parse(JSON.stringify(hit)) as CompressionResult;
|
||||
const cloned = JSON.parse(JSON.stringify(hit)) as CompressionResult;
|
||||
if (cloned.stats) {
|
||||
cloned.stats.memoHit = true;
|
||||
}
|
||||
return cloned;
|
||||
}
|
||||
|
||||
export function memoStore(key: string, result: CompressionResult): void {
|
||||
// Clone on STORE too (memoLookup already clones on read). Storing the caller's live
|
||||
// object would let a later mutation of it (e.g. an async engine holding a sub-ref)
|
||||
// corrupt the cached entry. Both ends isolated ⇒ the cache is immutable once stored.
|
||||
boundedSet(key, JSON.parse(JSON.stringify(result)) as CompressionResult);
|
||||
export function memoStore(key: string, result: CompressionResult): CompressionResult {
|
||||
// Clone on STORE (memoLookup also clones on read) so the caller's live object — which
|
||||
// an async engine may still hold a sub-ref to — cannot later corrupt the cached entry.
|
||||
// Returns the stored clone so callers that need a fresh instance (the common
|
||||
// `memoStore(key, result); return memoLookup(key)!` idiom) can avoid a redundant
|
||||
// second multi-MB deep clone of the body on the way out.
|
||||
const stored = JSON.parse(JSON.stringify(result)) as CompressionResult;
|
||||
boundedSet(key, stored);
|
||||
return stored;
|
||||
}
|
||||
|
||||
/** For tests only — clears the in-process memo store. */
|
||||
/** Observability stats for the in-process result memo store.
|
||||
* `windows` gives time-bucketed hit/miss/rate (1m/5m/15m/1h) so operators
|
||||
* see the *current* behavior during a spike, not the diluted lifetime rate.
|
||||
* `hits`/`misses`/`hitRate` remain the lifetime cumulative counters. */
|
||||
export function getMemoStats(): {
|
||||
size: number;
|
||||
capacity: number;
|
||||
hits: number;
|
||||
misses: number;
|
||||
hitRate: number;
|
||||
windows: {
|
||||
"1m": { hits: number; misses: number; hitRate: number };
|
||||
"5m": { hits: number; misses: number; hitRate: number };
|
||||
"15m": { hits: number; misses: number; hitRate: number };
|
||||
"1h": { hits: number; misses: number; hitRate: number };
|
||||
};
|
||||
} {
|
||||
const total = memoHits + memoMisses;
|
||||
return {
|
||||
size: memoMap.size,
|
||||
capacity: MEMO_CAP,
|
||||
hits: memoHits,
|
||||
misses: memoMisses,
|
||||
hitRate: total > 0 ? Math.round((memoHits / total) * 10000) / 100 : 0,
|
||||
windows: {
|
||||
"1m": windowStats(60_000),
|
||||
"5m": windowStats(5 * 60_000),
|
||||
"15m": windowStats(15 * 60_000),
|
||||
"1h": windowStats(60 * 60_000),
|
||||
},
|
||||
};
|
||||
}
|
||||
|
||||
/** For tests only — clears the in-process memo store and resets counters. */
|
||||
export function clearMemoStore(): void {
|
||||
memoMap.clear();
|
||||
lookupCountForTests = 0;
|
||||
memoHits = 0;
|
||||
memoMisses = 0;
|
||||
for (let i = 0; i < RING_CAP; i++) ring[i] = undefined;
|
||||
ringHead = 0;
|
||||
ringCount = 0;
|
||||
}
|
||||
export const resultMemoForTests = {
|
||||
get lookupCount(): number {
|
||||
|
||||
@@ -11,14 +11,22 @@ import {
|
||||
countTextTokens,
|
||||
isCodexTokenizerContext,
|
||||
tokenizerContextFromBody,
|
||||
MAX_EXACT_TOKEN_COUNT_CHARS,
|
||||
} from "../../../src/shared/utils/tiktokenCounter.ts";
|
||||
import {
|
||||
anthropicImageTokens,
|
||||
ANTHROPIC_IMAGE_BLOCK_OVERHEAD_TOKENS,
|
||||
openAIVisionTokens,
|
||||
} from "omniglyph";
|
||||
import { isInlineBase64ImageBlock } from "../contextManager.ts";
|
||||
import {
|
||||
jsonLength,
|
||||
jsonLengthStrippingBase64DataUris,
|
||||
rawLengthStrippingBase64DataUris,
|
||||
} from "../../utils/jsonSize.ts";
|
||||
|
||||
const CHARS_PER_TOKEN = 4;
|
||||
const DEFAULT_IMAGE_TOKEN_ESTIMATE = 1200;
|
||||
|
||||
/**
|
||||
* Anthropic image block shape this estimator recognizes:
|
||||
@@ -112,11 +120,15 @@ function decodePngDimensions(base64: string): { width: number; height: number }
|
||||
}
|
||||
}
|
||||
|
||||
/** Char-count fallback for one value (same accounting as the legacy estimator). */
|
||||
/** Char-count fallback for one value (using jsonLength to avoid allocating multi-MB strings).
|
||||
* Base64 data URIs embedded in arbitrary strings (not just structured image blocks) are
|
||||
* stripped so a tool-output screenshot doesn't inflate the token estimate (#7847 drift). */
|
||||
function charTokensOf(value: unknown): number {
|
||||
if (value === null || value === undefined) return 0;
|
||||
const str = typeof value === "string" ? value : JSON.stringify(value);
|
||||
return Math.ceil(str.length / CHARS_PER_TOKEN);
|
||||
if (typeof value === "string") {
|
||||
return Math.ceil(rawLengthStrippingBase64DataUris(value) / CHARS_PER_TOKEN);
|
||||
}
|
||||
return Math.ceil(jsonLengthStrippingBase64DataUris(value) / CHARS_PER_TOKEN);
|
||||
}
|
||||
|
||||
/**
|
||||
@@ -142,23 +154,42 @@ function blankImageBlocksAndSumImageTokens(body: Record<string, unknown>): {
|
||||
return content.map((block) => {
|
||||
if (isAnthropicPngImageBlock(block)) {
|
||||
const dims = decodePngDimensions(block.source.data);
|
||||
if (!dims) return block; // fall back to char-counting this block as-is
|
||||
if (!dims) {
|
||||
// Recognized image block that can't be decoded: use a bounded estimate rather
|
||||
// than char-counting the raw base64, which would inflate the token estimate
|
||||
// multi-MB (the #7847 OOM/drift class).
|
||||
imageTokens += DEFAULT_IMAGE_TOKEN_ESTIMATE;
|
||||
return { ...block, source: { ...block.source, data: "" } };
|
||||
}
|
||||
imageTokens += anthropicImageTokens(dims.width, dims.height, "standard");
|
||||
imageTokens += ANTHROPIC_IMAGE_BLOCK_OVERHEAD_TOKENS;
|
||||
return { ...block, source: { ...block.source, data: "" } };
|
||||
}
|
||||
if (isOpenAIChatPngImagePart(block)) {
|
||||
const dims = pngDimensionsFromDataUrl(block.image_url.url);
|
||||
if (!dims) return block;
|
||||
if (!dims) {
|
||||
imageTokens += DEFAULT_IMAGE_TOKEN_ESTIMATE;
|
||||
return { ...block, image_url: { ...block.image_url, url: "" } };
|
||||
}
|
||||
imageTokens += openAIVisionTokens(model, dims.width, dims.height);
|
||||
return { ...block, image_url: { ...block.image_url, url: "" } };
|
||||
}
|
||||
if (isOpenAIResponsesPngImagePart(block)) {
|
||||
const dims = pngDimensionsFromDataUrl(block.image_url);
|
||||
if (!dims) return block;
|
||||
if (!dims) {
|
||||
imageTokens += DEFAULT_IMAGE_TOKEN_ESTIMATE;
|
||||
return { ...block, image_url: "" };
|
||||
}
|
||||
imageTokens += openAIVisionTokens(model, dims.width, dims.height);
|
||||
return { ...block, image_url: "" };
|
||||
}
|
||||
if (isInlineBase64ImageBlock(block as Record<string, unknown>)) {
|
||||
// Inline-base64 image content-block shape (AI SDK / Gemini / flat) not
|
||||
// covered by the PNG decoders above. Keep the estimate bounded so a
|
||||
// multi-MB screenshot doesn't inflate the token count (#7847 drift).
|
||||
imageTokens += DEFAULT_IMAGE_TOKEN_ESTIMATE;
|
||||
return { ...block, image: "" };
|
||||
}
|
||||
return block;
|
||||
});
|
||||
};
|
||||
@@ -201,15 +232,19 @@ export function estimateCompressionTokens(text: string | object | null | undefin
|
||||
text as Record<string, unknown>
|
||||
);
|
||||
if (imageTokens === 0) {
|
||||
// Keep the legacy character estimate for generic payloads. Codex payloads use
|
||||
// the model-appropriate tokenizer so their compression stats match hard budgets.
|
||||
return useExactTokenizer
|
||||
? countTextTokens(JSON.stringify(text), tokenizerContext)
|
||||
: charTokensOf(text);
|
||||
// countTextTokens falls back to a char heuristic above MAX_EXACT_TOKEN_COUNT_CHARS,
|
||||
// so materializing JSON.stringify(text) for a large body would only allocate a
|
||||
// multi-MB transient that's immediately discarded (#7847 OOM class). Measure the
|
||||
// serialized length via jsonLength instead and skip the allocation when oversized.
|
||||
if (useExactTokenizer && jsonLength(text) <= MAX_EXACT_TOKEN_COUNT_CHARS) {
|
||||
return countTextTokens(JSON.stringify(text), tokenizerContext);
|
||||
}
|
||||
return charTokensOf(text);
|
||||
}
|
||||
return useExactTokenizer
|
||||
? countTextTokens(JSON.stringify(clone), tokenizerContext) + imageTokens
|
||||
: charTokensOf(clone) + imageTokens;
|
||||
if (useExactTokenizer && jsonLength(clone) <= MAX_EXACT_TOKEN_COUNT_CHARS) {
|
||||
return countTextTokens(JSON.stringify(clone), tokenizerContext) + imageTokens;
|
||||
}
|
||||
return charTokensOf(clone) + imageTokens;
|
||||
} catch {
|
||||
// Non-serializable/unexpected shape → fall back to the legacy char-count,
|
||||
// never throw out of an estimator.
|
||||
|
||||
@@ -331,6 +331,10 @@ function runCompression(
|
||||
...options,
|
||||
config: { ...options.config, memoizeCompressionResults: false },
|
||||
});
|
||||
// memoStore clones internally, so the cache entry stays isolated from the caller's
|
||||
// live object. Return the caller's own `result` (upstream #11727 semantics): handing
|
||||
// back the stored clone would let the caller's later mutations corrupt the cache —
|
||||
// the exact bug the result-memo mutation-isolation test guards.
|
||||
memoStore(key, result);
|
||||
return result;
|
||||
}
|
||||
@@ -564,6 +568,8 @@ async function runCompressionAsync(
|
||||
...options,
|
||||
config: { ...options.config, memoizeCompressionResults: false },
|
||||
});
|
||||
// Same contract as the sync path: store the internal clone; return the caller's own
|
||||
// object so later caller mutations cannot corrupt the cache (#11727 semantics).
|
||||
memoStore(key, result);
|
||||
return result;
|
||||
}
|
||||
|
||||
@@ -326,6 +326,8 @@ export interface CompressionStats {
|
||||
validationWarnings?: string[];
|
||||
validationErrors?: string[];
|
||||
fallbackApplied?: boolean;
|
||||
/** #7847 observability: true when this result was served from the result memo cache. */
|
||||
memoHit?: boolean;
|
||||
/**
|
||||
* Contabilidade física do OmniGlyph, normalizada pelo próprio pacote
|
||||
* (`normalizeAccounting`). Só número e enum — ver `omniglyphTelemetry.ts`
|
||||
|
||||
@@ -36,6 +36,10 @@ import {
|
||||
getResolvedModelCapabilities,
|
||||
supportsReasoning,
|
||||
} from "@/lib/modelCapabilities";
|
||||
import {
|
||||
jsonLengthStrippingBase64DataUris,
|
||||
rawLengthStrippingBase64DataUris,
|
||||
} from "../utils/jsonSize.ts";
|
||||
|
||||
// Effort → budget token mapping
|
||||
export const EFFORT_BUDGETS: Record<string, number> = {
|
||||
@@ -350,7 +354,8 @@ function applyAdaptiveBudget(body: unknown, cfg: Partial<ThinkingBudgetConfig>)
|
||||
const tools = Array.isArray(bodyRecord.tools) ? bodyRecord.tools : [];
|
||||
const toolCount = tools.length;
|
||||
|
||||
// Get last user message length
|
||||
// Get last user message length. Strip base64 data URIs so an inline image in the prompt
|
||||
// doesn't inflate lastMsgLength and silently bump the complexity multiplier.
|
||||
let lastMsgLength = 0;
|
||||
for (let i = messages.length - 1; i >= 0; i--) {
|
||||
const msg = messages[i];
|
||||
@@ -358,8 +363,8 @@ function applyAdaptiveBudget(body: unknown, cfg: Partial<ThinkingBudgetConfig>)
|
||||
if (msgRecord.role === "user") {
|
||||
lastMsgLength =
|
||||
typeof msgRecord.content === "string"
|
||||
? msgRecord.content.length
|
||||
: JSON.stringify(msgRecord.content || "").length;
|
||||
? rawLengthStrippingBase64DataUris(msgRecord.content)
|
||||
: jsonLengthStrippingBase64DataUris(msgRecord.content || "");
|
||||
break;
|
||||
}
|
||||
}
|
||||
|
||||
221
open-sse/utils/jsonHash.ts
Normal file
221
open-sse/utils/jsonHash.ts
Normal file
@@ -0,0 +1,221 @@
|
||||
import crypto from "node:crypto";
|
||||
|
||||
/**
|
||||
* Streaming JSON hash — computes `sha256hex(JSON.stringify(value))` WITHOUT
|
||||
* materializing the JSON string (#7847 OOM class). Several hot-path call sites
|
||||
* stringify a multi-megabyte request body just to hash it (compression memo
|
||||
* keys, cache keys). On a ~5 MiB agent body (with base64 screenshots) that
|
||||
* allocates a full ~5 MiB string, read once for a hash, then discarded.
|
||||
*
|
||||
* `jsonSha256()` walks the value and feeds the same bytes `JSON.stringify`
|
||||
* would emit directly into a `crypto.createHash("sha256")` stream, so peak
|
||||
* allocation stays bounded to a small rolling buffer.
|
||||
*
|
||||
* Semantics mirror `JSON.stringify` exactly:
|
||||
* - key order = `Object.keys()` order (insertion order)
|
||||
* - `undefined`/function/symbol object values drop the whole entry
|
||||
* - `undefined`/function/symbol array items render as `null`
|
||||
* - non-finite numbers render as `null`
|
||||
* - `BigInt` throws (matches JSON.stringify)
|
||||
* - Date / toJSON / non-plain containers fall back to `JSON.stringify` for
|
||||
* that subtree only (kept rare so big arrays stay on the fast path).
|
||||
*
|
||||
* Deterministic across calls: identical logical bodies always produce the
|
||||
* identical digest, so callers can replace `sha256hex(JSON.stringify(body))`
|
||||
* with `jsonSha256(body)` without changing cache/memo semantics.
|
||||
*/
|
||||
export function jsonSha256(value: unknown): string {
|
||||
const hash = crypto.createHash("sha256");
|
||||
writeValue(hash, value, new Set<object>());
|
||||
return hash.digest("hex");
|
||||
}
|
||||
|
||||
function isOmitted(value: unknown): boolean {
|
||||
return value === undefined || typeof value === "function" || typeof value === "symbol";
|
||||
}
|
||||
|
||||
function isPlainContainer(value: object): boolean {
|
||||
if (Array.isArray(value)) return true;
|
||||
const proto = Object.getPrototypeOf(value);
|
||||
return proto === Object.prototype || proto === null;
|
||||
}
|
||||
|
||||
function writeValue(
|
||||
hash: ReturnType<typeof crypto.createHash>,
|
||||
value: unknown,
|
||||
seen: Set<object>
|
||||
): void {
|
||||
if (writePrimitive(hash, value)) return;
|
||||
|
||||
const obj = value as object;
|
||||
// Date, Map, boxed primitives, class instances with toJSON — fall back to
|
||||
// JSON.stringify for THIS SUBTREE only, keeping multi-MB arrays on the
|
||||
// streaming path. JSON.stringify(Date) emits a quoted ISO string, so push
|
||||
// exactly the string form JSON.stringify would have produced.
|
||||
if (writeToJSONFallback(hash, obj)) return;
|
||||
|
||||
if (Array.isArray(obj)) {
|
||||
if (seen.has(obj)) {
|
||||
throw new TypeError("Converting circular structure to JSON");
|
||||
}
|
||||
seen.add(obj);
|
||||
try {
|
||||
writeArray(hash, obj, seen);
|
||||
} finally {
|
||||
seen.delete(obj);
|
||||
}
|
||||
} else {
|
||||
writePlainObject(hash, obj, seen);
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* toJSON / non-plain-container fallback: serializes the subtree with
|
||||
* JSON.stringify, exactly as JSON.stringify would have (undefined → the bare
|
||||
* token, e.g. an object-valued key being dropped later is not possible here
|
||||
* — writeValue callers already filter omissions). Returns true when handled.
|
||||
*/
|
||||
function writeToJSONFallback(
|
||||
hash: ReturnType<typeof crypto.createHash>,
|
||||
obj: object
|
||||
): boolean {
|
||||
const hasToJSON = typeof (obj as { toJSON?: unknown }).toJSON === "function";
|
||||
if (hasToJSON || !isPlainContainer(obj)) {
|
||||
const encoded = JSON.stringify(obj);
|
||||
hash.update(encoded === undefined ? "undefined" : encoded);
|
||||
return true;
|
||||
}
|
||||
return false;
|
||||
}
|
||||
|
||||
/** Writes JSON primitives and omissions. Returns true when `value` is fully handled. */
|
||||
function writePrimitive(hash: ReturnType<typeof crypto.createHash>, value: unknown): boolean {
|
||||
if (value === null) {
|
||||
hash.update("null");
|
||||
return true;
|
||||
}
|
||||
const type = typeof value;
|
||||
if (type === "string") {
|
||||
writeEncodedString(hash, value as string);
|
||||
return true;
|
||||
}
|
||||
if (type === "boolean") {
|
||||
hash.update(value ? "true" : "false");
|
||||
return true;
|
||||
}
|
||||
if (type === "number") {
|
||||
// Non-finite numbers serialize as null (matches JSON.stringify).
|
||||
hash.update(Number.isFinite(value as number) ? String(value) : "null");
|
||||
return true;
|
||||
}
|
||||
if (type === "bigint") {
|
||||
// Matches JSON.stringify, which throws rather than guessing an encoding.
|
||||
throw new TypeError("Do not know how to serialize a BigInt");
|
||||
}
|
||||
if (isOmitted(value) || type !== "object") {
|
||||
return true;
|
||||
}
|
||||
return false;
|
||||
}
|
||||
|
||||
function writeArray(
|
||||
hash: ReturnType<typeof crypto.createHash>,
|
||||
obj: unknown[],
|
||||
seen: Set<object>
|
||||
): void {
|
||||
hash.update("[");
|
||||
for (let i = 0; i < obj.length; i++) {
|
||||
if (i > 0) hash.update(",");
|
||||
const item = obj[i];
|
||||
if (isOmitted(item)) {
|
||||
hash.update("null"); // array items render as null
|
||||
} else {
|
||||
writeValue(hash, item, seen);
|
||||
}
|
||||
}
|
||||
hash.update("]");
|
||||
}
|
||||
|
||||
function writePlainObject(
|
||||
hash: ReturnType<typeof crypto.createHash>,
|
||||
obj: object,
|
||||
seen: Set<object>
|
||||
): void {
|
||||
if (seen.has(obj)) {
|
||||
throw new TypeError("Converting circular structure to JSON");
|
||||
}
|
||||
seen.add(obj);
|
||||
try {
|
||||
hash.update("{");
|
||||
let first = true;
|
||||
for (const key of Object.keys(obj)) {
|
||||
const item = (obj as Record<string, unknown>)[key];
|
||||
if (isOmitted(item)) continue; // entry disappears entirely
|
||||
if (!first) hash.update(",");
|
||||
first = false;
|
||||
writeEncodedString(hash, key);
|
||||
hash.update(":");
|
||||
writeValue(hash, item, seen);
|
||||
}
|
||||
hash.update("}");
|
||||
} finally {
|
||||
seen.delete(obj);
|
||||
}
|
||||
}
|
||||
|
||||
// Static escapes for fast paths: quote, backslash, and the short control
|
||||
// escapes JSON.stringify emits. Lookup avoids the escape ladder entirely.
|
||||
const SINGLE_ESCAPES = new Map<number, string>([
|
||||
[0x22, '\\"'],
|
||||
[0x5c, "\\\\"],
|
||||
[0x08, "\\b"],
|
||||
[0x09, "\\t"],
|
||||
[0x0a, "\\n"],
|
||||
[0x0c, "\\f"],
|
||||
[0x0d, "\\r"],
|
||||
]);
|
||||
|
||||
/** Writes one (possibly surrogate-paired) code unit's escaped form. */
|
||||
function appendEscapedChar(out: string[], value: string, i: number, code: number): number {
|
||||
const single = SINGLE_ESCAPES.get(code);
|
||||
if (single !== undefined) {
|
||||
out.push(single);
|
||||
return i;
|
||||
}
|
||||
if (code < 0x20) {
|
||||
out.push("\\u" + code.toString(16).padStart(4, "0"));
|
||||
return i;
|
||||
}
|
||||
if (code >= 0xd800 && code <= 0xdfff) {
|
||||
const next = i + 1 < value.length ? value.charCodeAt(i + 1) : NaN;
|
||||
const isHigh = code >= 0xd800 && code <= 0xdbff;
|
||||
if (isHigh && next >= 0xdc00 && next <= 0xdfff) {
|
||||
out.push(value[i] + value[i + 1]);
|
||||
return i + 1;
|
||||
}
|
||||
out.push("\\u" + code.toString(16).padStart(4, "0"));
|
||||
return i;
|
||||
}
|
||||
out.push(value[i]);
|
||||
return i;
|
||||
}
|
||||
|
||||
/** Writes a JSON-escaped, double-quoted string, flushing in ~8 KiB chunks. */
|
||||
function writeEncodedString(hash: ReturnType<typeof crypto.createHash>, value: string): void {
|
||||
const out: string[] = [];
|
||||
let buffered = 0;
|
||||
let i = 0;
|
||||
out.push('"');
|
||||
while (i < value.length) {
|
||||
const next = appendEscapedChar(out, value, i, value.charCodeAt(i));
|
||||
buffered += next - i + 1;
|
||||
i = next + 1;
|
||||
if (buffered > 8192) {
|
||||
hash.update(out.join(""));
|
||||
out.length = 0;
|
||||
buffered = 0;
|
||||
}
|
||||
}
|
||||
out.push('"');
|
||||
hash.update(out.join(""));
|
||||
}
|
||||
@@ -18,11 +18,14 @@
|
||||
* message history back onto the allocating path.
|
||||
*/
|
||||
|
||||
const BASE64_DATA_URI_RE = /data:image\/[a-z0-9.+-]+;base64,[A-Za-z0-9+/=]+/gi;
|
||||
|
||||
/** Length of a JSON-encoded string, including the surrounding quotes. */
|
||||
function encodedStringLength(value: string): number {
|
||||
function encodedStringLength(value: string, stripBase64 = false): number {
|
||||
const target = stripBase64 ? value.replace(BASE64_DATA_URI_RE, "") : value;
|
||||
let len = 2; // the quotes
|
||||
for (let i = 0; i < value.length; i++) {
|
||||
const code = value.charCodeAt(i);
|
||||
for (let i = 0; i < target.length; i++) {
|
||||
const code = target.charCodeAt(i);
|
||||
if (code === 0x22 || code === 0x5c) {
|
||||
len += 2; // \" and \\
|
||||
} else if (code === 0x08 || code === 0x09 || code === 0x0a || code === 0x0c || code === 0x0d) {
|
||||
@@ -33,7 +36,7 @@ function encodedStringLength(value: string): number {
|
||||
// Surrogates: a well-formed pair serializes as its two code units (2 chars); a LONE
|
||||
// surrogate is escaped as \uXXXX since ES2019 well-formed JSON.stringify.
|
||||
const isHigh = code <= 0xdbff;
|
||||
const next = isHigh ? value.charCodeAt(i + 1) : NaN;
|
||||
const next = isHigh ? target.charCodeAt(i + 1) : NaN;
|
||||
const paired = isHigh && next >= 0xdc00 && next <= 0xdfff;
|
||||
if (paired) {
|
||||
len += 2;
|
||||
@@ -66,14 +69,34 @@ function isPlainContainer(value: object): boolean {
|
||||
* Throws on circular structures and BigInt, exactly as JSON.stringify does.
|
||||
*/
|
||||
export function jsonLength(value: unknown): number {
|
||||
return lengthOf(value, new Set<object>());
|
||||
return lengthOf(value, new Set<object>(), false);
|
||||
}
|
||||
|
||||
function lengthOf(value: unknown, seen: Set<object>): number {
|
||||
/**
|
||||
* Same as `jsonLength`, but strips `data:image/*;base64,...` data URIs from strings
|
||||
* before counting, matching `countTextTokens(JSON.stringify(body))` semantics for
|
||||
* token heuristics without materializing the multi-megabyte string (#7847).
|
||||
*/
|
||||
export function jsonLengthStrippingBase64DataUris(value: unknown): number {
|
||||
return lengthOf(value, new Set<object>(), true);
|
||||
}
|
||||
|
||||
/**
|
||||
* Raw length of a string with `data:image/*;base64,...` data URIs removed. Unlike
|
||||
* `jsonLengthStrippingBase64DataUris`, this returns the plain code-unit count with NO
|
||||
* JSON-encoding overhead (no surrounding quotes/escaping). Use it where a threshold was
|
||||
* previously fed by `string.length` (e.g. thinking-budget complexity) but the value may
|
||||
* embed a base64 image.
|
||||
*/
|
||||
export function rawLengthStrippingBase64DataUris(value: string): number {
|
||||
return value.replace(BASE64_DATA_URI_RE, "").length;
|
||||
}
|
||||
|
||||
function lengthOf(value: unknown, seen: Set<object>, stripBase64: boolean): number {
|
||||
if (value === null) return 4; // "null"
|
||||
const type = typeof value;
|
||||
|
||||
if (type === "string") return encodedStringLength(value as string);
|
||||
if (type === "string") return encodedStringLength(value as string, stripBase64);
|
||||
if (type === "boolean") return value ? 4 : 5;
|
||||
if (type === "number") {
|
||||
// Non-finite numbers serialize as null.
|
||||
@@ -92,7 +115,8 @@ function lengthOf(value: unknown, seen: Set<object>): number {
|
||||
// Map, boxed primitives. Scoped to this subtree so the big arrays stay on the fast path.
|
||||
if (!isPlainContainer(obj) || typeof (obj as { toJSON?: unknown }).toJSON === "function") {
|
||||
const encoded = JSON.stringify(obj);
|
||||
return encoded === undefined ? 0 : encoded.length;
|
||||
if (encoded === undefined) return 0;
|
||||
return stripBase64 ? encoded.replace(BASE64_DATA_URI_RE, "").length : encoded.length;
|
||||
}
|
||||
|
||||
if (seen.has(obj)) {
|
||||
@@ -106,7 +130,7 @@ function lengthOf(value: unknown, seen: Set<object>): number {
|
||||
if (i > 0) len += 1; // comma
|
||||
const item = obj[i];
|
||||
// Omitted values render as null inside arrays rather than disappearing.
|
||||
len += isOmitted(item) ? 4 : lengthOf(item, seen);
|
||||
len += isOmitted(item) ? 4 : lengthOf(item, seen, stripBase64);
|
||||
}
|
||||
return len;
|
||||
}
|
||||
@@ -118,7 +142,7 @@ function lengthOf(value: unknown, seen: Set<object>): number {
|
||||
if (isOmitted(item)) continue; // the whole entry disappears
|
||||
if (!first) len += 1; // comma
|
||||
first = false;
|
||||
len += encodedStringLength(key) + 1 + lengthOf(item, seen); // "key":value
|
||||
len += encodedStringLength(key, false) + 1 + lengthOf(item, seen, stripBase64); // "key":value
|
||||
}
|
||||
return len;
|
||||
} finally {
|
||||
|
||||
@@ -1,6 +1,7 @@
|
||||
import { cloneLogPayload } from "@/lib/logPayloads";
|
||||
import { toNumber } from "@/shared/utils/numeric";
|
||||
import { FORMATS } from "../translator/formats.ts";
|
||||
import { jsonLength } from "./jsonSize.ts";
|
||||
|
||||
type StructuredSSEEvent = {
|
||||
index: number;
|
||||
@@ -914,7 +915,7 @@ export function createStructuredSSECollector(options: CollectorOptions = {}) {
|
||||
event.event = eventName;
|
||||
}
|
||||
|
||||
const serializedSize = JSON.stringify(event).length;
|
||||
const serializedSize = jsonLength(event);
|
||||
if (events.length >= maxEvents || usedBytes + serializedSize > maxBytes) {
|
||||
droppedEvents += 1;
|
||||
return;
|
||||
|
||||
@@ -25,6 +25,17 @@ curl https://localhost:20128/api/monitoring/health \
|
||||
-H "Authorization: Bearer $OMNIROUTE_TOKEN"
|
||||
```
|
||||
|
||||
### GET /api/monitoring/compression
|
||||
|
||||
Get compression result-memo statistics
|
||||
|
||||
In-process compression result-memo observability snapshot — size, capacity, lifetime hits/misses/hitRate plus 1m/5m/15m/1h windowed rates. Lightweight (no DB, no provider reads) companion to `GET /api/monitoring/health` intended for frequent polling. Sent with `Cache-Control: no-store, no-cache, must-revalidate`. Counters reset on process restart.
|
||||
|
||||
```bash
|
||||
curl https://localhost:20128/api/monitoring/compression \
|
||||
-H "Authorization: Bearer $OMNIROUTE_TOKEN"
|
||||
```
|
||||
|
||||
### GET /api/provider-metrics
|
||||
|
||||
GET provider metrics
|
||||
|
||||
45
src/app/api/monitoring/compression/route.ts
Normal file
45
src/app/api/monitoring/compression/route.ts
Normal file
@@ -0,0 +1,45 @@
|
||||
import { NextResponse } from "next/server";
|
||||
import pino from "pino";
|
||||
|
||||
const logger = pino({ name: "monitoring-compression-api" });
|
||||
|
||||
/**
|
||||
* GET /api/monitoring/compression — Compression result-memo observability snapshot
|
||||
*
|
||||
* Exposes the in-process compression result-memo stats (size, capacity, hits,
|
||||
* misses, hitRate) so the cache-hit efficiency of the memoized compression
|
||||
* path can be tracked over HTTP. This is the observability companion to the
|
||||
* #7847 OOM mitigations: a low memo hit rate on deterministic (lite/standard/
|
||||
* rtk) modes signals repeated full-pipeline re-runs that the cache was meant
|
||||
* to eliminate.
|
||||
*
|
||||
* Lightweight (no DB, no provider reads) and intentionally distinct from the
|
||||
* heavier /api/monitoring/health snapshot so it can be polled more frequently.
|
||||
*/
|
||||
export const dynamic = "force-dynamic";
|
||||
|
||||
export async function GET() {
|
||||
try {
|
||||
const { getMemoStats } = await import("@omniroute/open-sse/services/compression/index.ts");
|
||||
return NextResponse.json(
|
||||
{
|
||||
compression: {
|
||||
memo: getMemoStats(),
|
||||
},
|
||||
timestamp: new Date().toISOString(),
|
||||
},
|
||||
{
|
||||
status: 200,
|
||||
headers: {
|
||||
"Cache-Control": "no-store, no-cache, must-revalidate",
|
||||
},
|
||||
}
|
||||
);
|
||||
} catch (error) {
|
||||
logger.error({ err: error }, "GET /api/monitoring/compression failed");
|
||||
return NextResponse.json(
|
||||
{ status: "error", error: "compression_stats_unavailable" },
|
||||
{ status: 503 }
|
||||
);
|
||||
}
|
||||
}
|
||||
@@ -29,7 +29,7 @@ const encoders = new Map<TokenizerEncoding, Tiktoken>();
|
||||
* compression stats/estimates only, so a heuristic on oversized inputs is
|
||||
* acceptable and keeps the loop responsive.
|
||||
*/
|
||||
const MAX_EXACT_TOKEN_COUNT_CHARS = 50_000;
|
||||
export const MAX_EXACT_TOKEN_COUNT_CHARS = 50_000;
|
||||
|
||||
/**
|
||||
* Base64 data URIs (e.g. OpenAI-style `image_url.url`) must not be tokenized:
|
||||
|
||||
523
tests/unit/compression/memory-mitigations-edge-cases.test.ts
Normal file
523
tests/unit/compression/memory-mitigations-edge-cases.test.ts
Normal file
@@ -0,0 +1,523 @@
|
||||
/**
|
||||
* Comprehensive Edge Cases, Failure Modes, and Workflows Test Suite
|
||||
* for all #7847 OOM & Memory Mitigations in OmniRoute.
|
||||
*
|
||||
* Verifies the memory mitigations hold across edge cases and failure modes:
|
||||
* 1. jsonSha256: BigInt/circular throws, toJSON/Date, Unicode, control chars,
|
||||
* sparse arrays, undefined/function/symbol, deep nesting
|
||||
* 2. liveZone: fail-open on non-serializable messages, large base64 tool
|
||||
* output digests + frozen prefix reuse
|
||||
* 3. hardBudget: multimodal non-string content, already-in-budget, unreachable
|
||||
* budget, per-message proportional allocation
|
||||
* 4. thinkingBudget: adaptive multiplier scaling (messageCount/toolCount/lastMsg
|
||||
* length >2000) via the real applyThinkingBudget entry point
|
||||
* 5. stats.ts & codex engine: exact-vs-heuristic boundary and oversized bodies
|
||||
* 6. streamPayloadCollector: exact byte-limit accounting via jsonLength
|
||||
*/
|
||||
import { describe, it } from "node:test";
|
||||
import assert from "node:assert/strict";
|
||||
import crypto from "node:crypto";
|
||||
import { jsonSha256 } from "../../../open-sse/utils/jsonHash.ts";
|
||||
import { jsonLength } from "../../../open-sse/utils/jsonSize.ts";
|
||||
import { applyLiveZoneCompression } from "../../../open-sse/services/compression/liveZone.ts";
|
||||
import { applyHardBudget } from "../../../open-sse/services/compression/hardBudget.ts";
|
||||
import { applyThinkingBudget, ThinkingMode } from "../../../open-sse/services/thinkingBudget.ts";
|
||||
import { estimateCompressionTokens } from "../../../open-sse/services/compression/stats.ts";
|
||||
import { createStructuredSSECollector } from "../../../open-sse/utils/streamPayloadCollector.ts";
|
||||
import type { CompressionResult } from "../../../open-sse/services/compression/types.ts";
|
||||
import { adaptBodyForCompression } from "../../../open-sse/services/compression/bodyAdapter.ts";
|
||||
import { codexResponsesEngine } from "../../../open-sse/services/compression/engines/codexResponses/index.ts";
|
||||
|
||||
function sha256hex(text: string): string {
|
||||
return crypto.createHash("sha256").update(text).digest("hex");
|
||||
}
|
||||
|
||||
describe("Memory Mitigations — Comprehensive Edge Cases & Failure Modes", () => {
|
||||
// =========================================================================
|
||||
// 1. jsonSha256 Edge Cases & Failure Modes
|
||||
// =========================================================================
|
||||
describe("jsonSha256: Error handling & Edge cases", () => {
|
||||
it("throws TypeError on BigInt (matching JSON.stringify failure mode)", () => {
|
||||
assert.throws(
|
||||
() => jsonSha256({ val: BigInt(42) }),
|
||||
(err: unknown) => err instanceof TypeError
|
||||
);
|
||||
assert.throws(
|
||||
() => jsonSha256([1, 2, BigInt(99)]),
|
||||
(err: unknown) => err instanceof TypeError
|
||||
);
|
||||
});
|
||||
|
||||
it("throws TypeError on circular references (matching JSON.stringify)", () => {
|
||||
const circularObj: Record<string, unknown> = { a: 1 };
|
||||
circularObj.self = circularObj;
|
||||
assert.throws(
|
||||
() => jsonSha256(circularObj),
|
||||
(err: unknown) => err instanceof TypeError && /circular/i.test((err as Error).message)
|
||||
);
|
||||
|
||||
const circularArr: unknown[] = [1, 2];
|
||||
circularArr.push(circularArr);
|
||||
assert.throws(
|
||||
() => jsonSha256(circularArr),
|
||||
(err: unknown) => err instanceof TypeError && /circular/i.test((err as Error).message)
|
||||
);
|
||||
});
|
||||
|
||||
it("matches JSON.stringify hash for toJSON methods, Dates, and complex subtrees", () => {
|
||||
const custom = {
|
||||
name: "test",
|
||||
toJSON() {
|
||||
return { resolved: true, num: 123 };
|
||||
},
|
||||
};
|
||||
const date = new Date("2026-08-25T05:00:00.000Z");
|
||||
const complex = { item: custom, date, nested: [{ inside: custom }] };
|
||||
assert.equal(jsonSha256(complex), sha256hex(JSON.stringify(complex)));
|
||||
});
|
||||
|
||||
it("handles the full Unicode spectrum identically to JSON.stringify", () => {
|
||||
const unicodeCases = [
|
||||
"Hello 🌍 world 🚀",
|
||||
"👨👩👧👦 complex emoji sequence",
|
||||
"日本語のテストです。中文测试。한국어 테스트.",
|
||||
"∀x ∈ ℝ: x² ≥ 0 ∧ ∫ e^x dx = e^x + C",
|
||||
"Special quotes: „smart“ «guillemets» ‘single’ “double”",
|
||||
];
|
||||
for (const str of unicodeCases) {
|
||||
const payload = { text: str, arr: [str, { k: str }] };
|
||||
assert.equal(jsonSha256(payload), sha256hex(JSON.stringify(payload)));
|
||||
}
|
||||
});
|
||||
|
||||
it("handles control characters and escape sequences identically to JSON.stringify", () => {
|
||||
const controlCases = [
|
||||
"\x00\x01\x02\x03\x04\x05\x06\x07",
|
||||
"\b\t\n\x0b\f\r\x0e\x0f",
|
||||
"\x10\x11\x12\x13\x14\x15\x16\x17\x18\x19\x1a\x1b\x1c\x1d\x1e\x1f",
|
||||
'Quotes: " \\ and escaped \\" \\\\ \n \t',
|
||||
];
|
||||
for (const ctrl of controlCases) {
|
||||
const payload = { ctrl, nested: { value: ctrl } };
|
||||
assert.equal(jsonSha256(payload), sha256hex(JSON.stringify(payload)));
|
||||
}
|
||||
});
|
||||
|
||||
it("handles sparse arrays, undefined/function/symbol, NaN/Infinity identically", () => {
|
||||
const sparseArr = new Array(5);
|
||||
sparseArr[1] = "foo";
|
||||
sparseArr[3] = null;
|
||||
|
||||
const weirdObject = {
|
||||
a: undefined,
|
||||
b: () => {},
|
||||
c: Symbol("sym"),
|
||||
d: "kept",
|
||||
arr: [undefined, () => {}, Symbol("sym"), null, "kept", NaN, Infinity, -Infinity],
|
||||
};
|
||||
|
||||
assert.equal(jsonSha256(sparseArr), sha256hex(JSON.stringify(sparseArr)));
|
||||
assert.equal(jsonSha256(weirdObject), sha256hex(JSON.stringify(weirdObject)));
|
||||
});
|
||||
|
||||
it("handles deeply nested structures (depth 50) without recursion overflow", () => {
|
||||
let deep: Record<string, unknown> = { leaf: "value" };
|
||||
for (let i = 0; i < 50; i++) {
|
||||
deep = { level: i, next: deep };
|
||||
}
|
||||
assert.equal(jsonSha256(deep), sha256hex(JSON.stringify(deep)));
|
||||
});
|
||||
});
|
||||
|
||||
// =========================================================================
|
||||
// 2. liveZone Edge Cases & Failure Modes
|
||||
// =========================================================================
|
||||
describe("liveZone: Edge cases, failure modes & streaming digests", () => {
|
||||
it("fails open gracefully when a message contains non-serializable data", async () => {
|
||||
const circularContent: Record<string, unknown> = { role: "tool" };
|
||||
circularContent.self = circularContent;
|
||||
|
||||
const body = {
|
||||
messages: [{ role: "user", content: "hello" }, circularContent],
|
||||
};
|
||||
|
||||
let compressorCalled = false;
|
||||
const compressor = async (b: Record<string, unknown>) => {
|
||||
compressorCalled = true;
|
||||
return { body: b, compressed: false, stats: null };
|
||||
};
|
||||
|
||||
const result = await applyLiveZoneCompression(
|
||||
body,
|
||||
{ principalId: "p1", sessionId: "s1", variant: "v1" },
|
||||
compressor
|
||||
);
|
||||
|
||||
assert.ok(compressorCalled, "compressor called as fail-open fallback");
|
||||
assert.ok(result.body, "returned body intact");
|
||||
});
|
||||
|
||||
it("digests large base64 tool output and reuses frozen prefix on new user message", async () => {
|
||||
const largeBase64 = "C".repeat(2 * 1024 * 1024); // 2 MiB payload
|
||||
const body = {
|
||||
messages: [
|
||||
{ role: "user", content: "run tool" },
|
||||
{ role: "tool", content: largeBase64, tool_call_id: "call_123" },
|
||||
],
|
||||
};
|
||||
|
||||
let compressionRuns = 0;
|
||||
const compressor = async (b: Record<string, unknown>): Promise<CompressionResult> => {
|
||||
compressionRuns++;
|
||||
return {
|
||||
body: {
|
||||
...b,
|
||||
messages: (b.messages as Array<Record<string, unknown>>).map((m) =>
|
||||
m.role === "tool" ? { ...m, content: "compressed_tool" } : m
|
||||
),
|
||||
},
|
||||
compressed: true,
|
||||
stats: {
|
||||
originalTokens: 100,
|
||||
compressedTokens: 20,
|
||||
savingsPercent: 80,
|
||||
techniquesUsed: ["tool-compress"],
|
||||
mode: "stacked",
|
||||
timestamp: Date.now(),
|
||||
},
|
||||
};
|
||||
};
|
||||
|
||||
const opts = {
|
||||
principalId: "user_test",
|
||||
sessionId: "session_img",
|
||||
variant: "v1",
|
||||
ttlMinutes: 10,
|
||||
};
|
||||
const res1 = await applyLiveZoneCompression(body, opts, compressor);
|
||||
assert.equal(compressionRuns, 1, "first call ran compression and stored in liveZone");
|
||||
assert.equal(res1.compressed, true);
|
||||
|
||||
// Second request with same messages + 1 new user message reuses frozen tool output
|
||||
const body2 = {
|
||||
messages: [...body.messages, { role: "user", content: "what next?" }],
|
||||
};
|
||||
const res2 = await applyLiveZoneCompression(body2, opts, compressor);
|
||||
assert.ok(res2.body);
|
||||
const resMsgs = res2.body.messages as Array<Record<string, unknown>>;
|
||||
assert.equal(resMsgs.length, 3);
|
||||
assert.equal(resMsgs[1].content, "compressed_tool");
|
||||
});
|
||||
});
|
||||
|
||||
// =========================================================================
|
||||
// 3. hardBudget Edge Cases & Failure Modes
|
||||
// =========================================================================
|
||||
describe("hardBudget: Multimodal, boundary & warning failure modes", () => {
|
||||
it("preserves non-string multimodal content while compressing string content", () => {
|
||||
const body = {
|
||||
messages: [
|
||||
{ role: "system", content: "You are an assistant." },
|
||||
{
|
||||
role: "user",
|
||||
content: [
|
||||
{ type: "text", text: "Explain this diagram:" },
|
||||
{ type: "image", source: { type: "base64", data: "fakebase64" } },
|
||||
],
|
||||
},
|
||||
{
|
||||
role: "assistant",
|
||||
content: "This is a very long response that will be compressed. ".repeat(30),
|
||||
},
|
||||
],
|
||||
};
|
||||
|
||||
const result = applyHardBudget(body, { targetTokens: 40 });
|
||||
assert.ok(result.body);
|
||||
const msgs = result.body.messages as Array<Record<string, unknown>>;
|
||||
assert.equal(msgs.length, 3);
|
||||
// Non-string array content preserved intact (image block not dropped by token logic)
|
||||
assert.ok(Array.isArray(msgs[1].content));
|
||||
assert.equal((msgs[1].content as unknown[]).length, 2);
|
||||
assert.equal(result.compressed, true);
|
||||
assert.ok(result.stats);
|
||||
});
|
||||
|
||||
it("returns compressed:false when already within targetTokens", () => {
|
||||
const body = {
|
||||
messages: [{ role: "user", content: "Short message." }],
|
||||
};
|
||||
const result = applyHardBudget(body, { targetTokens: 1000 });
|
||||
assert.equal(result.compressed, false);
|
||||
assert.equal(result.stats, null);
|
||||
});
|
||||
|
||||
it("emits validationWarnings when preserved content prevents reaching target", () => {
|
||||
const body = {
|
||||
messages: [{ role: "user", content: "`preserve_code_block_that_exceeds_target`" }],
|
||||
};
|
||||
const result = applyHardBudget(body, { targetTokens: 1 });
|
||||
if (result.compressed) {
|
||||
assert.ok(
|
||||
result.stats?.validationWarnings?.some((w) => /could not reach target/i.test(w)),
|
||||
"expected a validationWarning when target unreachable"
|
||||
);
|
||||
}
|
||||
});
|
||||
});
|
||||
|
||||
// =========================================================================
|
||||
// 4. thinkingBudget Adaptive Multiplier via real entry point
|
||||
// =========================================================================
|
||||
describe("thinkingBudget: adaptive multiplier scaling", () => {
|
||||
function adaptiveBudgetFor(body: unknown, effort: string): number {
|
||||
const result = applyThinkingBudget(body, {
|
||||
mode: ThinkingMode.ADAPTIVE,
|
||||
effortLevel: effort,
|
||||
}) as { thinking?: { budget_tokens: number } };
|
||||
return result.thinking?.budget_tokens ?? 0;
|
||||
}
|
||||
|
||||
it("scales multiplier for long last user message (>2000 chars) on string content", () => {
|
||||
const shortBody = {
|
||||
model: "claude-opus-4-8",
|
||||
messages: [{ role: "user", content: "x".repeat(500) }],
|
||||
};
|
||||
const longBody = {
|
||||
model: "claude-opus-4-8",
|
||||
messages: [{ role: "user", content: "x".repeat(2500) }],
|
||||
};
|
||||
|
||||
const shortBudget = adaptiveBudgetFor(shortBody, "medium");
|
||||
const longBudget = adaptiveBudgetFor(longBody, "medium");
|
||||
|
||||
// Base 10240. Long last-msg adds +0.3 => ceil(10240*1.3) = 13312
|
||||
// (short stays at 1.0 => 10240, unless model caps).
|
||||
assert.equal(shortBudget, 10240);
|
||||
assert.ok(longBudget > shortBudget, `long budget ${longBudget} > short ${shortBudget}`);
|
||||
});
|
||||
|
||||
it("scales multiplier for >2000-char array content via jsonLength", () => {
|
||||
const longArrayBody = {
|
||||
model: "claude-opus-4-8",
|
||||
messages: [
|
||||
{
|
||||
role: "user",
|
||||
content: [
|
||||
{ type: "text", text: "x".repeat(1500) },
|
||||
{ type: "text", text: "y".repeat(1500) },
|
||||
],
|
||||
},
|
||||
],
|
||||
};
|
||||
const budget = adaptiveBudgetFor(longArrayBody, "medium");
|
||||
// array content length ~3000 > 2000 => multiplier 1.3
|
||||
assert.equal(budget, 10240 * 1.3);
|
||||
});
|
||||
|
||||
it("handles missing, empty, or null user messages without throwing", () => {
|
||||
const base: Record<string, unknown> = { model: "claude-opus-4-8", messages: [] };
|
||||
assert.doesNotThrow(() => adaptiveBudgetFor(base, "low"));
|
||||
assert.doesNotThrow(() =>
|
||||
adaptiveBudgetFor({ ...base, messages: [{ role: "system", content: "hi" }] }, "low")
|
||||
);
|
||||
assert.doesNotThrow(() =>
|
||||
adaptiveBudgetFor({ ...base, messages: [{ role: "user", content: "" }] }, "low")
|
||||
);
|
||||
assert.doesNotThrow(() =>
|
||||
adaptiveBudgetFor({ ...base, messages: [{ role: "user", content: null }] }, "low")
|
||||
);
|
||||
});
|
||||
});
|
||||
|
||||
// =========================================================================
|
||||
// 5. stats.ts exact-vs-heuristic boundary (50k chars)
|
||||
// =========================================================================
|
||||
describe("estimateCompressionTokens: boundary threshold behavior", () => {
|
||||
it("computes bounded estimates under and over the 50k-char threshold", () => {
|
||||
const underBoundary = {
|
||||
messages: [{ role: "user", content: "hello world ".repeat(3000) }], // ~36k chars
|
||||
};
|
||||
const overBoundary = {
|
||||
messages: [{ role: "user", content: "hello world ".repeat(6000) }], // ~72k chars
|
||||
};
|
||||
|
||||
const estUnder = estimateCompressionTokens(underBoundary);
|
||||
const estOver = estimateCompressionTokens(overBoundary);
|
||||
|
||||
assert.ok(estUnder > 0, "under-boundary estimate computed");
|
||||
assert.ok(estOver > 0, "over-boundary estimate computed");
|
||||
assert.ok(estOver > estUnder, "larger payload has larger token count");
|
||||
});
|
||||
|
||||
it("strips base64 data URIs embedded in arbitrary strings (tool-output screenshot)", () => {
|
||||
// A base64 screenshot embedded in a tool-output JSON *string* is NOT a structured
|
||||
// image block, but must not inflate the token estimate either. Prior to the fix,
|
||||
// charTokensOf counted the raw string length → ~200KB img inflated the estimate
|
||||
// (~52k tokens). Now the embedded data URI is stripped.
|
||||
const img = `data:image/png;base64,${"A".repeat(200 * 1024)}`;
|
||||
const body = {
|
||||
messages: [
|
||||
{ role: "user", content: "analyze the screenshot" },
|
||||
{
|
||||
role: "tool",
|
||||
content: JSON.stringify({ tool: "browser_snapshot", png: img, text: "dom" }),
|
||||
},
|
||||
],
|
||||
};
|
||||
const est = estimateCompressionTokens(body);
|
||||
assert.ok(est > 0, "estimate computed");
|
||||
assert.ok(
|
||||
est < 5000,
|
||||
`embedded 200KB data URI must be stripped, not inflate the estimate (got ${est})`
|
||||
);
|
||||
});
|
||||
});
|
||||
|
||||
// =========================================================================
|
||||
// 6. codexResponses oversized-body token guard (>50k chars → heuristic)
|
||||
// =========================================================================
|
||||
describe("codexResponses: oversized tool-output token estimate stays bounded", () => {
|
||||
it("compresses a >50k-char eligible tool output without inflating the token estimate", () => {
|
||||
// Pretty-printed JSON (>50k chars with whitespace to strip, but under the
|
||||
// 512KB maxCandidateBytes cap). minifyJson removes the whitespace so the
|
||||
// engine compresses it, and countCodexTokensForBody must engage the >50k
|
||||
// heuristic rather than a giant exact tokenizer pass.
|
||||
const pretty = Array.from({ length: 700 }, (_, i) => ({
|
||||
name: `src/module_${i}/file_${i}.ts`,
|
||||
status: "modified",
|
||||
meta: { lines: 40 + i, author: `dev_${i % 5}`, branch: "feature/compression" },
|
||||
note: "some descriptive content that gets minified away",
|
||||
}));
|
||||
const bigOutput = JSON.stringify(pretty, null, 2); // indented => minifiable
|
||||
|
||||
assert.ok(
|
||||
bigOutput.length > 50_000,
|
||||
`fixture must exceed 50k chars (got ${bigOutput.length})`
|
||||
);
|
||||
assert.ok(bigOutput.length < 512 * 1024, "fixture under maxCandidateBytes");
|
||||
|
||||
const adapter = adaptBodyForCompression({
|
||||
input: [
|
||||
{ type: "function_call", call_id: "c1", name: "run_command", arguments: "{}" },
|
||||
{ type: "function_call_output", call_id: "c1", output: bigOutput },
|
||||
],
|
||||
});
|
||||
const result = codexResponsesEngine.apply(adapter.body, {
|
||||
stepConfig: { enabled: true },
|
||||
});
|
||||
|
||||
assert.equal(result.compressed, true, "oversized eligible output should compress");
|
||||
assert.ok(result.stats, "stats present");
|
||||
// The token estimate must be bounded: a >50k-char body uses the heuristic
|
||||
// (jsonLength/4) rather than a full exact tokenizer, so it stays
|
||||
// proportional to the real content and never balloons.
|
||||
assert.ok(
|
||||
result.stats.originalTokens < bigOutput.length,
|
||||
"originalTokens bounded below raw char count"
|
||||
);
|
||||
assert.ok(result.stats.originalTokens > 0, "positive token estimate");
|
||||
assert.ok(
|
||||
result.stats.compressedTokens > 0 &&
|
||||
result.stats.compressedTokens <= result.stats.originalTokens
|
||||
);
|
||||
});
|
||||
|
||||
it("does not inflate originalTokens for oversized output embedding a base64 image", () => {
|
||||
// countCodexTokensForBody's oversized branch must strip base64 data URIs before the
|
||||
// char heuristic, matching countTextTokens(JSON.stringify(body)) semantics. Otherwise a
|
||||
// large embedded screenshot (~5x-10x raw length vs true tokens) inflates originalTokens
|
||||
// and distorts savingsPercent, the silent-threshold-drift class the review warned against.
|
||||
const imgBase64 = `data:image/png;base64,${"A".repeat(200 * 1024)}`;
|
||||
// Pretty-printed JSON (>50k chars) that minifyJson rewrites, so the engine produces stats.
|
||||
const bigOutput = JSON.stringify(
|
||||
{
|
||||
tool: "browser_snapshot",
|
||||
png: imgBase64,
|
||||
metadata: {
|
||||
url: "https://example.com/page",
|
||||
viewport: "1440x900",
|
||||
status: "complete",
|
||||
},
|
||||
text: "some surrounding snapshot text that should dominate the true token estimate",
|
||||
},
|
||||
null,
|
||||
2
|
||||
);
|
||||
assert.ok(
|
||||
bigOutput.length > 50_000,
|
||||
`fixture must exceed 50k chars (got ${bigOutput.length})`
|
||||
);
|
||||
|
||||
const adapter = adaptBodyForCompression({
|
||||
input: [
|
||||
{ type: "function_call", call_id: "c2", name: "browser_snapshot", arguments: "{}" },
|
||||
{ type: "function_call_output", call_id: "c2", output: bigOutput },
|
||||
],
|
||||
});
|
||||
const result = codexResponsesEngine.apply(adapter.body, { stepConfig: { enabled: true } });
|
||||
|
||||
assert.ok(result.stats, "stats present");
|
||||
// Without stripping, originalTokens ≈ (200KB base64 + overhead)/4 ≈ 51k+. With stripping,
|
||||
// it is proportional to the real text → well under 5k. Assert it stayed low.
|
||||
const raw = bigOutput.length;
|
||||
assert.ok(
|
||||
result.stats.originalTokens < raw / 4,
|
||||
`base64 must be stripped: originalTokens ${result.stats.originalTokens} should be well below raw/4 = ${raw / 4}`
|
||||
);
|
||||
assert.ok(
|
||||
result.stats.originalTokens < 5000,
|
||||
`embedded 200KB image must not inflate the estimate (got ${result.stats.originalTokens})`
|
||||
);
|
||||
assert.ok(result.stats.originalTokens > 0, "still a positive token estimate");
|
||||
});
|
||||
|
||||
it("does not allocate a giant exact tokenizer string for oversized non-string bodies", () => {
|
||||
// Non-tool, non-string message with a huge nested object still routes through
|
||||
// the jsonLength guard in countCodexTokensForBody (heuristic), not a huge exact stringify.
|
||||
const bigBlob = {
|
||||
data: Array.from({ length: 8000 }, (_, i) => ({ v: `chunk${i}_${"x".repeat(20)}` })),
|
||||
};
|
||||
const adapter = adaptBodyForCompression({ input: [bigBlob] });
|
||||
const result = codexResponsesEngine.apply(adapter.body, { stepConfig: { enabled: true } });
|
||||
// Should not throw and should not produce an inflated token count.
|
||||
assert.ok(result.body);
|
||||
assert.equal(result.compressed, false, "ineligible blob left untouched");
|
||||
});
|
||||
});
|
||||
|
||||
// =========================================================================
|
||||
// 7. streamPayloadCollector Exact Byte-Limit Accounting via jsonLength
|
||||
// =========================================================================
|
||||
describe("streamPayloadCollector: exact byte-limit accounting", () => {
|
||||
it("collects a bounded subset within maxBytes using jsonLength", () => {
|
||||
const collector = createStructuredSSECollector({
|
||||
maxEvents: 100,
|
||||
maxBytes: 200,
|
||||
});
|
||||
|
||||
collector.push({ role: "assistant", content: "hi" });
|
||||
assert.equal(collector.getEvents().length, 1);
|
||||
|
||||
for (let i = 0; i < 10; i++) {
|
||||
collector.push({ role: "assistant", content: `msg_${i}_${"x".repeat(30)}` });
|
||||
}
|
||||
|
||||
const events = collector.getEvents();
|
||||
assert.ok(events.length >= 1 && events.length < 10, `bounded events ${events.length}`);
|
||||
const totalBytes = events.reduce((sum, e) => sum + jsonLength(e), 0);
|
||||
assert.ok(totalBytes <= 200, `totalBytes ${totalBytes} <= maxBytes 200`);
|
||||
});
|
||||
|
||||
it("does not exceed maxEvents even when individual events are tiny", () => {
|
||||
const collector = createStructuredSSECollector({
|
||||
maxEvents: 5,
|
||||
maxBytes: 100000,
|
||||
});
|
||||
for (let i = 0; i < 20; i++) {
|
||||
collector.push({ role: "assistant", content: `m${i}` });
|
||||
}
|
||||
assert.equal(collector.getEvents().length, 5);
|
||||
});
|
||||
});
|
||||
});
|
||||
176
tests/unit/compression/oom-memo-memory.test.ts
Normal file
176
tests/unit/compression/oom-memo-memory.test.ts
Normal file
@@ -0,0 +1,176 @@
|
||||
/**
|
||||
* E2E memory probe for the #7847 OOM mitigations, exercised through the REAL
|
||||
* public entry point `applyCompression` (strategySelector) — not a mock.
|
||||
*
|
||||
* Drives the memoized deterministic path (mode "lite", principalId set) with a
|
||||
* realistic multi-MB base64 image payload. The mitigations under test eliminate
|
||||
* throwaway multi-MB `JSON.stringify(body)` / deep-clone transients in exactly
|
||||
* this path (streaming makeMemoKey hash, memoStore single-clone return).
|
||||
*
|
||||
* Run:
|
||||
* node --import tsx/esm --import ./open-sse/utils/setupPolyfill.ts \
|
||||
* --test --test-force-exit tests/unit/compression/oom-memo-memory.test.ts
|
||||
*/
|
||||
import { describe, it } from "node:test";
|
||||
import assert from "node:assert/strict";
|
||||
import { applyCompression } from "../../../open-sse/services/compression/strategySelector.ts";
|
||||
import {
|
||||
makeMemoKey,
|
||||
memoStore,
|
||||
clearMemoStore,
|
||||
getMemoStats,
|
||||
} from "../../../open-sse/services/compression/resultMemo.ts";
|
||||
import type { CompressionResult } from "../../../open-sse/services/compression/types.ts";
|
||||
|
||||
function anonHeapMb(): number {
|
||||
// V8 heap used + external array buffers: the transient-allocation class the
|
||||
// OOM report tracked. Repeatable in-process proxy (not exact RSS).
|
||||
const m = process.memoryUsage();
|
||||
return (m.heapUsed + m.arrayBuffers) / (1024 * 1024);
|
||||
}
|
||||
|
||||
function base64Body(mb: number): Record<string, unknown> {
|
||||
const block = "A".repeat(Math.round(mb * 1024 * 1024 * 0.75)); // ~4:3 base64
|
||||
return {
|
||||
model: "claude-sonnet-4-5",
|
||||
messages: [
|
||||
{ role: "user", content: "analyze this screenshot" },
|
||||
{
|
||||
role: "user",
|
||||
content: [
|
||||
{ type: "image", source: { type: "base64", media_type: "image/png", data: block } },
|
||||
],
|
||||
},
|
||||
// Collapsible whitespace so the lite engine actually runs (stats non-null).
|
||||
{ role: "user", content: "word1 word2 word3\n\n\n\nword4" },
|
||||
],
|
||||
};
|
||||
}
|
||||
|
||||
const liteConfig = {
|
||||
enabled: true,
|
||||
defaultMode: "lite",
|
||||
memoizeCompressionResults: true,
|
||||
lite: { compressToolResults: true },
|
||||
engines: {} as Record<string, unknown>,
|
||||
};
|
||||
|
||||
describe("oom-memo e2e: public applyCompression path with large base64 payload", () => {
|
||||
it("runs memoized lite compression on a ~3MiB body without runaway allocation", () => {
|
||||
clearMemoStore();
|
||||
const body = base64Body(3);
|
||||
const principal = "e2e-principal";
|
||||
const opts = {
|
||||
config: liteConfig as never,
|
||||
principalId: principal,
|
||||
model: "claude-sonnet-4-5",
|
||||
supportsVision: true,
|
||||
};
|
||||
|
||||
const gc = (globalThis as { gc?: () => void }).gc;
|
||||
const before = anonHeapMb();
|
||||
const result = applyCompression(body, "lite", opts);
|
||||
// Large array buffers may need more than one forced cycle to release.
|
||||
if (gc) for (let i = 0; i < 3; i++) gc();
|
||||
const after = anonHeapMb();
|
||||
|
||||
// Compression actually ran (didn't bail to no-op) and returned valid stats.
|
||||
assert.ok(result.body, "compression returned a body");
|
||||
assert.equal(result.stats!.mode, "lite");
|
||||
|
||||
// Token estimate bounded (not base64-inflated ~1.35M).
|
||||
const est = result.stats!.originalTokens;
|
||||
assert.ok(est > 0 && est < 10_000, `estimate ${est} should be bounded, not base64-inflated`);
|
||||
|
||||
// Identical body + principal ⇒ memoized cache hit (identity preserved).
|
||||
const hit = applyCompression(body, "lite", opts);
|
||||
assert.deepEqual(hit.body, result.body, "memoized cache hit returns identical body");
|
||||
assert.equal(hit.stats!.originalTokens, result.stats!.originalTokens);
|
||||
assert.equal(hit.stats!.memoHit, true, "cache hit is observable via stats.memoHit");
|
||||
|
||||
// Memo observability counters reflect the hit.
|
||||
const memo = getMemoStats();
|
||||
assert.ok(memo.hits >= 1, `expected >=1 memo hit, got ${memo.hits}`);
|
||||
assert.ok(memo.misses >= 1, "first call was a miss");
|
||||
assert.equal(memo.size, 1, "one memoized entry held");
|
||||
assert.ok(memo.hitRate > 0, "hit rate reported");
|
||||
assert.ok(memo.capacity >= memo.size, "size within capacity");
|
||||
|
||||
// Windowed stats: this fresh run produced exactly 1 miss + 1 hit, so the
|
||||
// 1m window must report hitRate=50 with hits=1/misses=1 (windows reflect
|
||||
// *current* traffic, not a diluted all-time rate).
|
||||
assert.equal(memo.windows["1m"].hits, 1, "1m window counts the hit");
|
||||
assert.equal(memo.windows["1m"].misses, 1, "1m window counts the miss");
|
||||
assert.equal(memo.windows["1m"].hitRate, 50, "1m window hit rate is 50%");
|
||||
for (const w of ["5m", "15m", "1h"] as const) {
|
||||
assert.equal(memo.windows[w].hits, 1, `${w} window counts the hit`);
|
||||
assert.equal(memo.windows[w].misses, 1, `${w} window counts the miss`);
|
||||
}
|
||||
|
||||
// Retained heap after the full hot path must not have ballooned by the body
|
||||
// size (old double-clone pinned ~2x body transient). Generous headroom.
|
||||
// Without --expose-gc (CI shard runner), heapUsed can still momentarily
|
||||
// hold GC-pending transients, so the retained-heap assertion is only
|
||||
// meaningful when forced collection is available.
|
||||
const retained = after - before;
|
||||
if (gc) {
|
||||
assert.ok(
|
||||
retained < 30,
|
||||
`retained heap grew ${retained.toFixed(1)} MiB after 3MiB body (>30MiB = uncollected transient)`
|
||||
);
|
||||
}
|
||||
|
||||
// Streaming memo key is deterministic and principal-scoped.
|
||||
const k1 = makeMemoKey(body, "lite", liteConfig as never, principal, "claude-sonnet-4-5", true);
|
||||
const k2 = makeMemoKey(
|
||||
{ ...body },
|
||||
"lite",
|
||||
liteConfig as never,
|
||||
principal,
|
||||
"claude-sonnet-4-5",
|
||||
true
|
||||
);
|
||||
assert.equal(k1, k2);
|
||||
const k3 = makeMemoKey(
|
||||
body,
|
||||
"lite",
|
||||
liteConfig as never,
|
||||
"e2e-other",
|
||||
"claude-sonnet-4-5",
|
||||
true
|
||||
);
|
||||
assert.notEqual(k1, k3);
|
||||
});
|
||||
|
||||
it("memoStore single-clone return is isolated from the caller's live object", () => {
|
||||
clearMemoStore();
|
||||
const body = base64Body(1);
|
||||
const key = "k-" + Math.random().toString(36).slice(2);
|
||||
const messages = body.messages as Array<Record<string, unknown>>;
|
||||
const result: CompressionResult = {
|
||||
body,
|
||||
compressed: true,
|
||||
stats: {
|
||||
originalTokens: 5,
|
||||
compressedTokens: 4,
|
||||
savingsPercent: 20,
|
||||
techniquesUsed: ["lite"],
|
||||
mode: "lite",
|
||||
timestamp: Date.now(),
|
||||
},
|
||||
};
|
||||
const stored = memoStore(key, result);
|
||||
assert.notEqual(stored, result, "store returns a clone, not the live object");
|
||||
assert.notEqual(stored.body, result.body, "body is deep-cloned");
|
||||
assert.equal(
|
||||
(stored.body.messages as unknown[]).length,
|
||||
(result.body.messages as unknown[]).length
|
||||
);
|
||||
messages.push({ role: "user", content: "must not leak" });
|
||||
assert.equal(
|
||||
(stored.body.messages as unknown[]).length,
|
||||
3,
|
||||
"cache entry unaffected by caller mutation"
|
||||
);
|
||||
});
|
||||
});
|
||||
86
tests/unit/json-hash.test.ts
Normal file
86
tests/unit/json-hash.test.ts
Normal file
@@ -0,0 +1,86 @@
|
||||
import { describe, it } from "node:test";
|
||||
import assert from "node:assert/strict";
|
||||
import crypto from "node:crypto";
|
||||
import { jsonSha256 } from "../../open-sse/utils/jsonHash.ts";
|
||||
|
||||
function sha256hex(text: string): string {
|
||||
return crypto.createHash("sha256").update(text).digest("hex");
|
||||
}
|
||||
|
||||
describe("jsonSha256 matches sha256hex(JSON.stringify(x)) for serializable values", () => {
|
||||
const cases: Array<unknown> = [
|
||||
null,
|
||||
0,
|
||||
1,
|
||||
-1,
|
||||
3.14159,
|
||||
NaN,
|
||||
Infinity,
|
||||
-Infinity,
|
||||
true,
|
||||
false,
|
||||
"",
|
||||
"plain",
|
||||
'with "quotes" and \\backslash',
|
||||
"line\nbreak\ttab\rcr\bbs\fform",
|
||||
"\u0000\u001f control chars",
|
||||
"emoji 🚀 and surrogate \ud83d\ude00",
|
||||
"unpaired \ud800 lone",
|
||||
"mixed \ud83d\ude00\u0041\uD800X",
|
||||
[],
|
||||
[1, 2, 3],
|
||||
[[1], [2], [3]],
|
||||
[undefined, null, 1, "x"],
|
||||
{},
|
||||
{ a: 1, b: "two", c: [true, false] },
|
||||
{ z: 1, a: 2, m: 3 }, // insertion order preserved
|
||||
{ nested: { deep: { deeper: [{ ok: 1 }, null] } } },
|
||||
{ fn: () => 1, ignored: undefined, kept: "x" }, // omitted keys
|
||||
["http://x", { url: "http://y" }],
|
||||
{
|
||||
model: "gpt-4o",
|
||||
messages: [
|
||||
{
|
||||
role: "user",
|
||||
content: [
|
||||
{ type: "text", text: "hi" },
|
||||
{
|
||||
type: "image_url",
|
||||
image_url: { url: "data:image/png;base64," + "A".repeat(5_400_000) },
|
||||
},
|
||||
],
|
||||
},
|
||||
],
|
||||
},
|
||||
{
|
||||
// iBrowse MCP local-image shape with a large raw base64 payload.
|
||||
messages: [
|
||||
{
|
||||
role: "user",
|
||||
content: [{ type: "image", data: "A".repeat(5_400_000), mimeType: "image/png" }],
|
||||
},
|
||||
],
|
||||
},
|
||||
];
|
||||
|
||||
for (const value of cases) {
|
||||
const label =
|
||||
typeof value === "string" && value.length > 40
|
||||
? `string(${value.length})`
|
||||
: JSON.stringify(value)?.slice(0, 50);
|
||||
it(`matches for ${label}`, () => {
|
||||
const expected = sha256hex(JSON.stringify(value));
|
||||
assert.equal(jsonSha256(value), expected);
|
||||
});
|
||||
}
|
||||
|
||||
it("throws on BigInt like JSON.stringify", () => {
|
||||
assert.throws(() => jsonSha256({ n: 1n }), TypeError);
|
||||
});
|
||||
|
||||
it("throws on circular structures like JSON.stringify", () => {
|
||||
const obj: Record<string, unknown> = { a: 1 };
|
||||
obj.self = obj;
|
||||
assert.throws(() => jsonSha256(obj), TypeError);
|
||||
});
|
||||
});
|
||||
Reference in New Issue
Block a user