From 9a33cdac1d5840eff1cf64d91c0a0259792e39d9 Mon Sep 17 00:00:00 2001 From: Chirag Singhal <76880977+chirag127@users.noreply.github.com> Date: Tue, 7 Jul 2026 08:23:52 +0530 Subject: [PATCH] fix(compression): add intra-message dedup to session-dedup engine (#6467) (#6501) Intra-message dedup for the session-dedup compression engine (#6467) + fallbackReason surfacing + fusion rate-limit detail. Integrated into release/v3.8.46 with a TDD regression. --- CHANGELOG.md | 1 + open-sse/services/compression/diffHelper.ts | 11 +++ .../engines/session-dedup/index.ts | 67 +++++++++++++++++-- open-sse/services/fusion.ts | 16 ++++- src/app/api/compression/preview/route.ts | 2 + .../session-dedup-intra-message-6467.test.ts | 56 ++++++++++++++++ 6 files changed, 146 insertions(+), 7 deletions(-) create mode 100644 tests/unit/compression/session-dedup-intra-message-6467.test.ts diff --git a/CHANGELOG.md b/CHANGELOG.md index 26463bd75b..a0e5bea5fd 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -31,6 +31,7 @@ ### 🐛 Bug Fixes +- **fix(compression):** the session-dedup engine now also deduplicates a large multi-line block repeated **within a single message** (intra-message dedup), not just across turns; the compression-preview API surfaces a `fallbackReason`, and the fusion panel reports how many models were rate-limited vs failed on total-panel failure (#6501 — thanks @chirag127). - **fix(compression):** a stacked-pipeline step naming an unregistered engine now surfaces a `validationErrors` entry instead of silently no-op'ing, so misconfigured pipelines are visible in the preview API (#6506 — thanks @chirag127). - **feat(usage):** add a **Codex reset-credit redemption** flow to the Provider Limits UI — a `useCodexResetCreditRedemption` hook + `/api/usage/codex-reset-credit` route + `codexResetCredits` lib let you redeem banked Codex reset credits from the quota card. Regression guard: `tests/unit/codex-reset-credits.test.ts`. (thanks @JxnLexn) - **feat(glm):** add **team-plan quota settings** for `glm-cn` connections — a dedicated `GlmTeamQuotaFields` form section (team quota id / limits) threaded through the Add/Edit connection modals, persisted via `providerSpecificData`, with the GLM usage service reading the team quota. Regression guards: `tests/unit/glm-team-quota.test.ts`, `provider-specific-data-schema.test.ts`. (thanks @hao3039032) diff --git a/open-sse/services/compression/diffHelper.ts b/open-sse/services/compression/diffHelper.ts index aef855c118..6652b05602 100644 --- a/open-sse/services/compression/diffHelper.ts +++ b/open-sse/services/compression/diffHelper.ts @@ -28,6 +28,7 @@ export interface CompressionPreviewDiff { validationWarnings: string[]; validationErrors: string[]; fallbackApplied: boolean; + fallbackReason?: string; heatmap?: CompressionHeatmap; } @@ -187,6 +188,15 @@ export function buildCompressionPreviewDiff( ? [{ type: "same", text: "[diff omitted: input too large]" }] : buildCompressionDiff(original, compressed); + let fallbackReason: string | undefined; + if (validation.fallbackApplied) { + fallbackReason = validation.errors.length > 0 + ? `validation-failed: ${validation.errors[0]}` + : "validation-failed"; + } else if (stats?.fallbackApplied) { + fallbackReason = "compression-fallback"; + } + const result: CompressionPreviewDiff = { segments, preservedBlocks: preserved, @@ -198,6 +208,7 @@ export function buildCompressionPreviewDiff( ], validationErrors: [...(stats?.validationErrors ?? []), ...validation.errors], fallbackApplied: Boolean(stats?.fallbackApplied || validation.fallbackApplied), + ...(fallbackReason && { fallbackReason }), }; if (heatmapMode) { diff --git a/open-sse/services/compression/engines/session-dedup/index.ts b/open-sse/services/compression/engines/session-dedup/index.ts index 4a2b2303c3..306b79561f 100644 --- a/open-sse/services/compression/engines/session-dedup/index.ts +++ b/open-sse/services/compression/engines/session-dedup/index.ts @@ -88,6 +88,53 @@ function findSuffixBlocks( // ─── two-pass dedup on message texts ───────────────────────────────────────── +/** + * Deduplicates repeated lines within a single message (intra-message dedup). + * Replaces repeated suffix blocks with markers. + */ +function dedupeWithinMessage( + text: string, + minBlockChars: number +): { deduped: string; changed: boolean } { + const lines = text.split("\n"); + const blocks = findSuffixBlocks(lines, minBlockChars); + + if (blocks.length < 2) return { deduped: text, changed: false }; + + // Find the most common block (likely candidate for intra-message dedup). + const blockFreq = new Map(); + for (const { block } of blocks) { + blockFreq.set(block, (blockFreq.get(block) || 0) + 1); + } + + // Sort by frequency descending, then by length descending (prefer replacing more common, longer blocks first). + const sortedBlocks = [...blocks].sort((a, b) => { + const freqDiff = (blockFreq.get(b.block) || 0) - (blockFreq.get(a.block) || 0); + return freqDiff !== 0 ? freqDiff : b.block.length - a.block.length; + }); + + let result = text; + let changed = false; + + 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; + if (occurrences < 2) continue; + + const sha = hashBlock(block); + const marker = `[dedup:ref sha=${sha}]`; + // Replace ALL occurrences except the first (keep the original once). + let count = 0; + result = result.replace(new RegExp(block.replace(/[.*+?^${}()|[\]\\]/g, "\\$&"), "g"), () => { + count++; + return count === 1 ? block : marker; + }); + changed = true; + } + + return { deduped: result, changed }; +} + /** * Runs two-pass dedup over an ordered list of (msgIdx, text) pairs. * Returns the replaced texts for duplicate messages, a reverse map, and a count. @@ -99,6 +146,21 @@ function dedupMessageTexts( deduped: Map; dedupCount: number; } { + const deduped = new Map(); + let dedupCount = 0; + + // Single-message case: apply intra-message dedup. + if (msgTexts.length === 1) { + const { text, msgIdx } = msgTexts[0]; + const { deduped: dedupedText, changed } = dedupeWithinMessage(text, minBlockChars); + if (changed) { + deduped.set(msgIdx, dedupedText); + dedupCount++; + } + return { deduped, dedupCount }; + } + + // Multi-message case: apply cross-turn dedup. // Pass 1: for each message, extract suffix blocks and record first ownership. // `firstSeen`: sha → { ownerMsgIdx, block } const firstSeen = new Map(); @@ -115,9 +177,6 @@ function dedupMessageTexts( } // Pass 2: for each message, find blocks that were FIRST seen in an earlier message. - const deduped = new Map(); - let dedupCount = 0; - for (const { msgIdx, text } of msgTexts) { const lines = text.split("\n"); const blocks = findSuffixBlocks(lines, minBlockChars); @@ -202,7 +261,7 @@ function processMessages( } } - if (msgTexts.length < 2) { + if (msgTexts.length === 0) { return { messages, dedupCount: 0 }; } diff --git a/open-sse/services/fusion.ts b/open-sse/services/fusion.ts index 2ebde14a3a..5d6f5bac0a 100644 --- a/open-sse/services/fusion.ts +++ b/open-sse/services/fusion.ts @@ -268,6 +268,7 @@ export async function handleFusionChat({ // 2. Collect successful answers. const answers: Array<{ model: string; text: string }> = []; + const rateLimited: string[] = []; for (let i = 0; i < settled.length; i++) { const res = settled[i]; const model = panel[i]; @@ -288,7 +289,12 @@ export async function handleFusionChat({ } const resp = res as Response; if (!resp.ok) { - log.warn("FUSION", `Panel ${model} failed`, { status: resp.status }); + if (resp.status === 429) { + rateLimited.push(model); + log.warn("FUSION", `Panel ${model} rate-limited`, { status: resp.status }); + } else { + log.warn("FUSION", `Panel ${model} failed`, { status: resp.status }); + } continue; } try { @@ -309,8 +315,12 @@ export async function handleFusionChat({ // 3. Degrade gracefully when the panel is too thin to fuse. if (answers.length === 0) { - log.warn("FUSION", "All panel models failed"); - return errorResponse(503, "All fusion panel models failed"); + const detail = + rateLimited.length > 0 + ? `${rateLimited.length} models rate-limited, ${panel.length - rateLimited.length} failed` + : `all ${panel.length} models failed`; + log.warn("FUSION", `No live models: ${detail}`); + return errorResponse(503, `All fusion panel models failed (${detail})`); } if (answers.length === 1) { log.info( diff --git a/src/app/api/compression/preview/route.ts b/src/app/api/compression/preview/route.ts index 00e6d8481c..3153298ffa 100644 --- a/src/app/api/compression/preview/route.ts +++ b/src/app/api/compression/preview/route.ts @@ -256,10 +256,12 @@ export async function POST(req: Request) { errors: diff.validationErrors, warnings: diff.validationWarnings, fallbackApplied: diff.fallbackApplied, + ...(diff.fallbackReason && { fallbackReason: diff.fallbackReason }), }, validationWarnings: diff.validationWarnings, validationErrors: diff.validationErrors, fallbackApplied: diff.fallbackApplied, + ...(diff.fallbackReason && { fallbackReason: diff.fallbackReason }), ...(diff.heatmap ? { heatmap: diff.heatmap } : {}), }); } catch (err: unknown) { diff --git a/tests/unit/compression/session-dedup-intra-message-6467.test.ts b/tests/unit/compression/session-dedup-intra-message-6467.test.ts new file mode 100644 index 0000000000..5ceee4bc23 --- /dev/null +++ b/tests/unit/compression/session-dedup-intra-message-6467.test.ts @@ -0,0 +1,56 @@ +/** + * #6467 — intra-message dedup for the session-dedup compression engine. + * + * Before this change the engine bailed out whenever a single message was present + * (`msgTexts.length < 2`), so a lone message that repeats a large multi-line block + * inside itself was never deduplicated. The fix adds a single-message path that + * replaces the repeated block with a `[dedup:ref sha=…]` marker (keeping the first + * occurrence). + * + * Run: node --import tsx/esm --test tests/unit/compression/session-dedup-intra-message-6467.test.ts + */ + +import { describe, it } from "node:test"; +import assert from "node:assert/strict"; + +import { sessionDedupEngine } from "../../../open-sse/services/compression/engines/session-dedup/index.ts"; + +// A block that comfortably clears both thresholds: ≥ MIN_BLOCK_LINES (3) lines and +// ≥ minBlockChars (80) chars. +const BLOCK = `function expensiveCalc(x) { + const a = x * 2; + const b = a + 100; + const c = b * b - 7; + return c; +}`; + +function singleMessageBody(repeats: number): Record { + const parts: string[] = []; + for (let i = 0; i < repeats; i++) { + parts.push(`--- section ${i} ---`); + parts.push(BLOCK); + } + return { messages: [{ role: "user", content: parts.join("\n") }] }; +} + +describe("session-dedup intra-message dedup (#6467)", () => { + it("deduplicates a block repeated within a single message", () => { + const body = singleMessageBody(3); + const result = sessionDedupEngine.apply(body, { stepConfig: {} }); + + assert.equal(result.compressed, true, "a single message with an internal repeat must compress"); + const text = (result.body.messages as Array<{ content: string }>)[0].content; + assert.match(text, /\[dedup:ref sha=/, "repeated occurrences must be replaced by a ref marker"); + // The first occurrence is kept verbatim; the block must still be recoverable once. + assert.ok(text.includes(BLOCK), "the first occurrence of the block must be preserved"); + // 3 occurrences → 1 kept + 2 markers. + const markerCount = (text.match(/\[dedup:ref sha=/g) || []).length; + assert.equal(markerCount, 2, "two of the three occurrences must become markers"); + }); + + it("leaves a single message with no internal repetition untouched", () => { + const body = singleMessageBody(1); + const result = sessionDedupEngine.apply(body, { stepConfig: {} }); + assert.equal(result.compressed, false, "no repeated block → no compression"); + }); +});