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

@@ -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)

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) {

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<string, number>();
+  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<number, string>;
   dedupCount: number;
 } {
+  const deduped = new Map<number, string>();
+  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<string, { ownerMsgIdx: number; block: string }>();
@@ -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<number, string>();
-  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 };
   }
 

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(

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) {

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<string, unknown> {
+  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");
+  });
+});