Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
69 changes: 59 additions & 10 deletions open-sse/services/compression/engines/session-dedup/index.ts
Original file line number Diff line number Diff line change
Expand Up @@ -46,6 +46,15 @@ const ENGINE_ID = "session-dedup";
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.
*/
const MAX_SUFFIX_WORK_CHARS = 32 * 1024 * 1024;
const SUFFIX_WORK_BUDGET_WARNING = "session-dedup: skipped (suffix work budget exceeded)";

type SuffixWorkBudget = { remaining: number };

// ─── hash helper (SHA-256 prefix, collision-resistant) ───────────────────────

Expand All @@ -58,6 +67,24 @@ 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;
}
Comment thread
RaviTharuma marked this conversation as resolved.
Resolved

/**
* 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
Expand Down Expand Up @@ -118,7 +145,9 @@ 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);
Expand Down Expand Up @@ -240,7 +269,7 @@ type MessageLike = {
function processMessages(
messages: MessageLike[],
minBlockChars: number
): { messages: MessageLike[]; dedupCount: number } {
): { messages: MessageLike[]; dedupCount: number; suffixWorkBudgetExceeded: boolean } {
// Collect (msgIdx, text) for non-system string-content messages.
// For multipart, index each text part separately.
const msgTexts: Array<{ msgIdx: number; text: string }> = [];
Expand All @@ -262,13 +291,24 @@ function processMessages(
}

if (msgTexts.length === 0) {
return { messages, dedupCount: 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 };
}
}

const { deduped, dedupCount } = dedupMessageTexts(msgTexts, minBlockChars);

if (dedupCount === 0) {
return { messages, dedupCount: 0 };
return { messages, dedupCount: 0, suffixWorkBudgetExceeded: false };
}

const result = messages.map((msg, i) => {
Expand Down Expand Up @@ -297,7 +337,7 @@ function processMessages(
return { ...msg };
});

return { messages: result, dedupCount };
return { messages: result, dedupCount, suffixWorkBudgetExceeded: false };
}

// ─── schema & validation ──────────────────────────────────────────────────────
Expand Down Expand Up @@ -343,7 +383,8 @@ 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");
}
Expand Down Expand Up @@ -394,10 +435,18 @@ export const sessionDedupEngine: CompressionEngine = {
}

const start = performance.now();
const { messages: exactMessages, dedupCount } = processMessages(
messages as MessageLike[],
minBlockChars
);
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: finalMessages, fuzzyCount } = runFuzzyPass(
exactMessages,
Expand Down
32 changes: 32 additions & 0 deletions tests/fixtures/compression/session-dedup-memory-7849.ts
Original file line number Diff line number Diff line change
@@ -0,0 +1,32 @@
import { applyStackedCompression } from "../../../open-sse/services/compression/index.ts";

const lines = Array.from({ length: 4_000 }, (_, index) => {
const prefix = `line-${index.toString().padStart(4, "0")}:`;
return prefix + "x".repeat(80 - prefix.length);
});
const body = {
messages: [{ role: "tool", content: lines.join("\n") }],
};
const enginesRun: string[] = [];
const result = applyStackedCompression(
body,
[
{ engine: "session-dedup" },
{ engine: "lite" },
{ engine: "rtk" },
{ engine: "headroom" },
{ engine: "caveman" },
],
{
onEngineStep: (step) => {
enginesRun.push(step.engine);
},
}
);

process.stdout.write(
JSON.stringify({
enginesRun,
warnings: result.stats?.validationWarnings ?? [],
})
);
138 changes: 138 additions & 0 deletions tests/unit/compression/session-dedup-memory-7849.test.ts
Original file line number Diff line number Diff line change
@@ -0,0 +1,138 @@
import assert from "node:assert/strict";
import { spawnSync } from "node:child_process";
import { dirname, join } from "node:path";
import test from "node:test";
import { fileURLToPath } from "node:url";

import { sessionDedupEngine } from "../../../open-sse/services/compression/engines/session-dedup/index.ts";

const REPO_ROOT = join(dirname(fileURLToPath(import.meta.url)), "../../..");
const FIXTURE = join(REPO_ROOT, "tests/fixtures/compression/session-dedup-memory-7849.ts");
const SUFFIX_WORK_BUDGET = 32 * 1024 * 1024;
const SUFFIX_WORK_BUDGET_WARNING = "session-dedup: skipped (suffix work budget exceeded)";

function makeFixedWidthText(lineCount: number, lineChars: number, tag: string): string {
return Array.from({ length: lineCount }, (_, index) => {
const prefix = `${tag}-${index.toString().padStart(4, "0")}:`;
assert.ok(prefix.length <= lineChars);
return prefix + "x".repeat(lineChars - prefix.length);
}).join("\n");
}

function projectedSuffixWork(text: string, passCount: number): number {
let work = 0;
for (let start = 0; start <= text.length; start++) {
if (start === 0 || text.charCodeAt(start - 1) === 10) {
work += (text.length - start) * passCount;
}
}
return work;
}

function makeSharedBudgetBody(): Record<string, unknown> {
return {
messages: [
{ role: "tool", content: makeFixedWidthText(600, 49, "first") },
{ role: "tool", content: makeFixedWidthText(600, 49, "second") },
],
};
}

test("#7849: shares the two-pass suffix-work budget across all messages", () => {
const body = makeSharedBudgetBody();
const messages = body.messages as Array<{ content: string }>;
const perMessageWork = messages.map(({ content }) => projectedSuffixWork(content, 2));

assert.ok(
perMessageWork.every((work) => work < SUFFIX_WORK_BUDGET),
"each message must fit the two-pass budget on its own"
);
assert.ok(
messages.reduce((total, { content }) => total + projectedSuffixWork(content, 1), 0) <
SUFFIX_WORK_BUDGET,
"the pair must fit if incorrectly charged for only one pass"
);
assert.ok(
perMessageWork.reduce((total, work) => total + work, 0) > SUFFIX_WORK_BUDGET,
"the pair must exceed the shared budget when correctly charged for two passes"
);

for (const message of messages) {
const individualResult = sessionDedupEngine.apply({
messages: [message, { role: "assistant", content: "a unique short companion" }],
});
assert.equal(individualResult.stats, null, "each message must be accepted individually");
}

const result = sessionDedupEngine.apply(body);
assert.deepEqual(result.stats?.validationWarnings, [SUFFIX_WORK_BUDGET_WARNING]);
});

test("#7849: exhausted suffix-work budget fails open with exact zero-savings stats", () => {
const body = makeSharedBudgetBody();
const result = sessionDedupEngine.apply(body);

assert.strictEqual(result.body, body, "budget exhaustion must return the input body by identity");
assert.equal(result.compressed, false);
assert.ok(result.stats, "budget exhaustion must return explanatory stats");
assert.equal(result.stats.originalTokens, result.stats.compressedTokens);
assert.equal(result.stats.savingsPercent, 0);
assert.deepEqual(result.stats.validationWarnings, [SUFFIX_WORK_BUDGET_WARNING]);
});

test("#7849: near-boundary under-budget request still deduplicates", () => {
const repeatedText = makeFixedWidthText(578, 49, "same");
const projectedWork = projectedSuffixWork(repeatedText, 2) * 2;
assert.ok(projectedWork <= SUFFIX_WORK_BUDGET);
assert.ok(
SUFFIX_WORK_BUDGET - projectedWork < 100_000,
"fixture must remain close to the work-budget boundary"
);

const body = {
messages: [
{ role: "user", content: repeatedText },
{ role: "user", content: repeatedText },
],
};
const result = sessionDedupEngine.apply(body);
const messages = result.body.messages as Array<{ content: string }>;

assert.equal(result.compressed, true);
assert.equal(messages[0].content, repeatedText);
assert.match(messages[1].content, /^\[dedup:ref sha=[0-9a-f]{24}\]$/);
assert.ok((result.stats?.savingsPercent ?? 0) > 0);
assert.deepEqual(result.stats?.validationWarnings ?? [], []);
});

test(
"#7849: line-rich long context stays within a 512 MiB heap and the stacked pipeline continues",
{ timeout: 60_000 },
() => {
const child = spawnSync(
process.execPath,
["--max-old-space-size=512", "--import", "tsx/esm", FIXTURE],
{
cwd: REPO_ROOT,
encoding: "utf8",
timeout: 45_000,
}
);

assert.equal(
child.status,
0,
`compression child must not OOM or time out\nstdout: ${child.stdout}\nstderr: ${child.stderr}`
);

const output = JSON.parse(child.stdout) as {
enginesRun: string[];
warnings: string[];
};
assert.deepEqual(output.enginesRun, ["session-dedup", "lite", "rtk", "headroom", "caveman"]);
assert.ok(
output.warnings.includes("session-dedup: skipped (suffix work budget exceeded)"),
`expected an explicit session-dedup work-budget warning, got ${JSON.stringify(output.warnings)}`
);
}
);
Loading